🎖️GitЯра🎖️
Commit 41414f832f938b655c97bc07509bc140ea16e2a3
Parents : e9d09a3
Author : Jeremiah K <17190268+jeremiah-k@users.noreply.github.com>
Signature : Signature validation error
Date : 2026-08-15T23:49:38Z
Committer : GitHub <noreply@github.com>
Date : 2026-08-15T23:49:38Z
fix(discovery): restore radio state after interrupted scans (#6717)
Changes
24 files changed, 3965 insertions(+), 712 deletions(-)
Diff
diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/DiscoveryDao.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/DiscoveryDao.kt
index 0632f12ce6..38ecfc4633 100644
--- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/DiscoveryDao.kt
+++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/DiscoveryDao.kt
@@ -25,6 +25,7 @@ import kotlinx.coroutines.flow.Flow
import org.meshtastic.core.database.entity.DiscoveredNodeEntity
import org.meshtastic.core.database.entity.DiscoveryPresetResultEntity
import org.meshtastic.core.database.entity.DiscoverySessionEntity
+import org.meshtastic.core.database.entity.DiscoverySessionStatus
@Dao
@Suppress("TooManyFunctions")
@@ -46,25 +47,60 @@ interface DiscoveryDao {
@Query("SELECT * FROM discovery_session WHERE id = :sessionId")
suspend fun getSession(sessionId: Long): DiscoverySessionEntity?
+ @Query("UPDATE discovery_session SET completion_status = :status WHERE id = :sessionId")
+ suspend fun updateSessionCompletionStatus(sessionId: Long, status: String): Int
+
+ @Query(
+ "UPDATE discovery_session SET total_unique_nodes = :totalUniqueNodes, " +
+ "total_dwell_seconds = :totalDwellSeconds, total_messages = :totalMessages, " +
+ "total_sensor_packets = :totalSensorPackets, " +
+ "furthest_node_distance = :furthestNodeDistance, avg_channel_utilization = :avgChannelUtilization " +
+ "WHERE id = :sessionId",
+ )
+ suspend fun updateSessionAggregates(
+ sessionId: Long,
+ totalUniqueNodes: Int,
+ totalDwellSeconds: Long,
+ totalMessages: Int,
+ totalSensorPackets: Int,
+ furthestNodeDistance: Double,
+ avgChannelUtilization: Double,
+ )
+
+ @Query(
+ "UPDATE discovery_session SET completion_status = :status WHERE id = :sessionId " +
+ "AND completion_status IN (" +
+ DiscoverySessionStatus.RECOVERABLE_SQL_LIST +
+ ")",
+ )
+ suspend fun updateRecoverableSessionCompletionStatus(sessionId: Long, status: String): Int
+
@Query("SELECT * FROM discovery_session WHERE id = :sessionId")
fun getSessionFlow(sessionId: Long): Flow<DiscoverySessionEntity?>
@Query("DELETE FROM discovery_session WHERE id = :sessionId")
suspend fun deleteSession(sessionId: Long)
- @Query("UPDATE discovery_session SET completion_status = 'interrupted' WHERE completion_status = 'in_progress'")
+ @Query(
+ "UPDATE discovery_session SET completion_status = '" +
+ DiscoverySessionStatus.INTERRUPTED +
+ "' WHERE completion_status = '" +
+ DiscoverySessionStatus.IN_PROGRESS +
+ "'",
+ )
suspend fun markInterruptedSessions()
/**
- * The most recent session left mid-scan by a prior process (crash, BLE loss, or force-quit) for [deviceAddress] —
- * "in_progress" if the process died before [markInterruptedSessions] ever ran, "interrupted" otherwise.
+ * The most recent recoverable session for [deviceAddress]. This includes active/interrupted scans and terminal
+ * scans whose home-radio restoration is still pending, so callers must inspect completionStatus rather than
+ * assuming an unfinished scan.
*/
@Query(
- """
- SELECT * FROM discovery_session
- WHERE device_address = :deviceAddress AND completion_status IN ('in_progress', 'interrupted')
- ORDER BY timestamp DESC LIMIT 1
- """,
+ "SELECT * FROM discovery_session " +
+ "WHERE device_address = :deviceAddress " +
+ "AND completion_status IN (" +
+ DiscoverySessionStatus.RECOVERABLE_SQL_LIST +
+ ") ORDER BY timestamp DESC LIMIT 1",
)
suspend fun getInterruptedSession(deviceAddress: String): DiscoverySessionEntity?
diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/SwitchingDiscoveryDao.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/SwitchingDiscoveryDao.kt
index 07a137cd66..b60aab4f55 100644
--- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/SwitchingDiscoveryDao.kt
+++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/SwitchingDiscoveryDao.kt
@@ -32,8 +32,8 @@ import org.meshtastic.core.database.entity.DiscoverySessionEntity
* barrier (a mid-scan merge can't snapshot-then-retire the DB underneath an in-flight session write). A callback is
* never replayed after it starts, so higher layers must make any retry policy explicit where idempotency is known.
*
- * `withDb` only returns null when no database is open, which [DatabaseProvider] guarantees can't happen (the default DB
- * is the floor), so the non-null coercions below are structural, not behavioral.
+ * `withDb` can return null while no database is available. Collection reads use empty defaults, count reads return
+ * zero, and operations that require a created row retain non-null checks so callers cannot proceed with an invalid id.
*/
@Suppress("TooManyFunctions")
class SwitchingDiscoveryDao(private val dbManager: DatabaseProvider) : DiscoveryDao {
@@ -56,6 +56,35 @@ class SwitchingDiscoveryDao(private val dbManager: DatabaseProvider) : Discovery
override suspend fun getSession(sessionId: Long): DiscoverySessionEntity? =
dbManager.withDb { it.discoveryDao().getSession(sessionId) }
+ override suspend fun updateSessionCompletionStatus(sessionId: Long, status: String): Int =
+ dbManager.withDb { it.discoveryDao().updateSessionCompletionStatus(sessionId, status) } ?: 0
+
+ override suspend fun updateSessionAggregates(
+ sessionId: Long,
+ totalUniqueNodes: Int,
+ totalDwellSeconds: Long,
+ totalMessages: Int,
+ totalSensorPackets: Int,
+ furthestNodeDistance: Double,
+ avgChannelUtilization: Double,
+ ) {
+ dbManager.withDb {
+ it.discoveryDao()
+ .updateSessionAggregates(
+ sessionId,
+ totalUniqueNodes,
+ totalDwellSeconds,
+ totalMessages,
+ totalSensorPackets,
+ furthestNodeDistance,
+ avgChannelUtilization,
+ )
+ }
+ }
+
+ override suspend fun updateRecoverableSessionCompletionStatus(sessionId: Long, status: String): Int =
+ dbManager.withDb { it.discoveryDao().updateRecoverableSessionCompletionStatus(sessionId, status) } ?: 0
+
override fun getSessionFlow(sessionId: Long): Flow<DiscoverySessionEntity?> =
dbManager.observeCurrentDb { it.discoveryDao().getSessionFlow(sessionId) }
diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/entity/DiscoverySessionEntity.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/entity/DiscoverySessionEntity.kt
index 2536396735..fb9982794e 100644
--- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/entity/DiscoverySessionEntity.kt
+++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/entity/DiscoverySessionEntity.kt
@@ -33,7 +33,8 @@ data class DiscoverySessionEntity(
@ColumnInfo(name = "total_messages", defaultValue = "0") val totalMessages: Int = 0,
@ColumnInfo(name = "total_sensor_packets", defaultValue = "0") val totalSensorPackets: Int = 0,
@ColumnInfo(name = "furthest_node_distance", defaultValue = "0.0") val furthestNodeDistance: Double = 0.0,
- @ColumnInfo(name = "completion_status", defaultValue = "'complete'") val completionStatus: String = "complete",
+ @ColumnInfo(name = "completion_status", defaultValue = "'" + DiscoverySessionStatus.COMPLETE + "'")
+ val completionStatus: String = DiscoverySessionStatus.COMPLETE,
@ColumnInfo(name = "ai_summary") val aiSummary: String? = null,
@ColumnInfo(name = "user_latitude", defaultValue = "0.0") val userLatitude: Double = 0.0,
@ColumnInfo(name = "user_longitude", defaultValue = "0.0") val userLongitude: Double = 0.0,
diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/entity/DiscoverySessionStatus.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/entity/DiscoverySessionStatus.kt
new file mode 100644
index 0000000000..7421cfa959
--- /dev/null
+++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/entity/DiscoverySessionStatus.kt
@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.database.entity
+
+/** Persisted discovery-session states shared by database queries, scan execution, and recovery. */
+object DiscoverySessionStatus {
+ const val IN_PROGRESS = "in_progress"
+ const val INTERRUPTED = "interrupted"
+ const val COMPLETE = "complete"
+ const val FAILED = "failed"
+ const val STOPPED = "stopped"
+ const val RESTORED = "restored"
+ const val UNRESTORABLE = "unrestorable"
+ const val RESTORE_PENDING_COMPLETE = "restore_pending_complete"
+ const val RESTORE_PENDING_FAILED = "restore_pending_failed"
+ const val RESTORE_PENDING_STOPPED = "restore_pending_stopped"
+
+ /** SQL literal list required by Room's compile-time query parser for sessions that reconnect can still recover. */
+ const val RECOVERABLE_SQL_LIST =
+ "'$IN_PROGRESS', '$INTERRUPTED', '$RESTORE_PENDING_COMPLETE', " +
+ "'$RESTORE_PENDING_FAILED', '$RESTORE_PENDING_STOPPED'"
+
+ /** In-memory counterpart of [RECOVERABLE_SQL_LIST]. A unit test keeps the two declarations synchronized. */
+ val RECOVERABLE: Set<String> =
+ linkedSetOf(IN_PROGRESS, INTERRUPTED, RESTORE_PENDING_COMPLETE, RESTORE_PENDING_FAILED, RESTORE_PENDING_STOPPED)
+}
diff --git a/core/database/src/commonTest/kotlin/org/meshtastic/core/database/dao/CommonDiscoveryDaoTest.kt b/core/database/src/commonTest/kotlin/org/meshtastic/core/database/dao/CommonDiscoveryDaoTest.kt
index 119eb11d1e..4b0f6ca5d4 100644
--- a/core/database/src/commonTest/kotlin/org/meshtastic/core/database/dao/CommonDiscoveryDaoTest.kt
+++ b/core/database/src/commonTest/kotlin/org/meshtastic/core/database/dao/CommonDiscoveryDaoTest.kt
@@ -24,6 +24,7 @@ import org.meshtastic.core.database.MeshtasticDatabase
import org.meshtastic.core.database.entity.DiscoveredNodeEntity
import org.meshtastic.core.database.entity.DiscoveryPresetResultEntity
import org.meshtastic.core.database.entity.DiscoverySessionEntity
+import org.meshtastic.core.database.entity.DiscoverySessionStatus
import org.meshtastic.core.database.getInMemoryDatabaseBuilder
import kotlin.test.AfterTest
import kotlin.test.Test
@@ -43,11 +44,18 @@ abstract class CommonDiscoveryDaoTest {
@AfterTest
fun closeDb() {
- database.close()
+ if (::database.isInitialized) database.close()
}
// region Session CRUD
+ @Test
+ fun recoverableSetMatchesSqlList() = runTest {
+ val fromSql =
+ DiscoverySessionStatus.RECOVERABLE_SQL_LIST.split(",").map { it.trim().removeSurrounding("'") }.toSet()
+ assertEquals(fromSql, DiscoverySessionStatus.RECOVERABLE)
+ }
+
@Test
fun insertSession_returnsAutoGeneratedId() = runTest {
createDb()
@@ -84,6 +92,78 @@ abstract class CommonDiscoveryDaoTest {
assertEquals(5, updated.totalUniqueNodes)
}
+ @Test
+ fun updateSessionAggregates_preservesCompletionStatus() = runTest {
+ createDb()
+ val id =
+ dao.insertSession(
+ testSession(homePreset = "LONG_FAST").copy(completionStatus = DiscoverySessionStatus.COMPLETE),
+ )
+
+ dao.updateSessionAggregates(
+ sessionId = id,
+ totalUniqueNodes = 5,
+ totalDwellSeconds = 12L,
+ totalMessages = 3,
+ totalSensorPackets = 4,
+ furthestNodeDistance = 500.0,
+ avgChannelUtilization = 22.5,
+ )
+
+ val updated = dao.getSession(id)!!
+ assertEquals(DiscoverySessionStatus.COMPLETE, updated.completionStatus)
+ assertEquals(5, updated.totalUniqueNodes)
+ assertEquals(12L, updated.totalDwellSeconds)
+ assertEquals(3, updated.totalMessages)
+ assertEquals(4, updated.totalSensorPackets)
+ assertEquals(500.0, updated.furthestNodeDistance)
+ assertEquals(22.5, updated.avgChannelUtilization)
+ }
+
+ @Test
+ fun updateSessionCompletionStatus_updatesOnlyTheStatus() = runTest {
+ createDb()
+ val id = dao.insertSession(testSession(homePreset = "LONG_FAST"))
+
+ assertEquals(1, dao.updateSessionCompletionStatus(id, DiscoverySessionStatus.FAILED))
+ assertEquals(0, dao.updateSessionCompletionStatus(Long.MAX_VALUE, DiscoverySessionStatus.FAILED))
+
+ val updated = dao.getSession(id)!!
+ assertEquals(DiscoverySessionStatus.FAILED, updated.completionStatus)
+ assertEquals("LONG_FAST", updated.homePreset)
+ }
+
+ @Test
+ fun updateRecoverableSessionCompletionStatus_updatesRecoverableSession() = runTest {
+ createDb()
+ val id = dao.insertSession(testSession().copy(completionStatus = DiscoverySessionStatus.RESTORE_PENDING_FAILED))
+
+ val updatedRows = dao.updateRecoverableSessionCompletionStatus(id, DiscoverySessionStatus.FAILED)
+
+ assertEquals(1, updatedRows)
+ assertEquals(DiscoverySessionStatus.FAILED, dao.getSession(id)!!.completionStatus)
+ }
+
+ @Test
+ fun updateRecoverableSessionCompletionStatus_ignoresTerminalSession() = runTest {
+ createDb()
+ val id = dao.insertSession(testSession().copy(completionStatus = DiscoverySessionStatus.COMPLETE))
+
+ val updatedRows = dao.updateRecoverableSessionCompletionStatus(id, DiscoverySessionStatus.FAILED)
+
+ assertEquals(0, updatedRows)
+ assertEquals(DiscoverySessionStatus.COMPLETE, dao.getSession(id)!!.completionStatus)
+ }
+
+ @Test
+ fun updateRecoverableSessionCompletionStatus_returnsZeroForMissingSession() = runTest {
+ createDb()
+
+ val updatedRows = dao.updateRecoverableSessionCompletionStatus(999L, DiscoverySessionStatus.FAILED)
+
+ assertEquals(0, updatedRows)
+ }
+
@Test
fun deleteSession_removesRow() = runTest {
createDb()
@@ -99,32 +179,65 @@ abstract class CommonDiscoveryDaoTest {
@Test
fun getInterruptedSession_returnsInProgressSessionForDevice() = runTest {
createDb()
- dao.insertSession(testSession().copy(deviceAddress = "x:AA:BB:CC:DD:EE:FF", completionStatus = "in_progress"))
+ dao.insertSession(
+ testSession()
+ .copy(deviceAddress = "x:AA:BB:CC:DD:EE:FF", completionStatus = DiscoverySessionStatus.IN_PROGRESS),
+ )
val found = dao.getInterruptedSession("x:AA:BB:CC:DD:EE:FF")
assertNotNull(found)
- assertEquals("in_progress", found.completionStatus)
+ assertEquals(DiscoverySessionStatus.IN_PROGRESS, found.completionStatus)
}
@Test
fun getInterruptedSession_returnsInterruptedSessionToo() = runTest {
createDb()
- dao.insertSession(testSession().copy(deviceAddress = "x:AA:BB:CC:DD:EE:FF", completionStatus = "interrupted"))
+ dao.insertSession(
+ testSession()
+ .copy(deviceAddress = "x:AA:BB:CC:DD:EE:FF", completionStatus = DiscoverySessionStatus.INTERRUPTED),
+ )
val found = dao.getInterruptedSession("x:AA:BB:CC:DD:EE:FF")
assertNotNull(found)
- assertEquals("interrupted", found.completionStatus)
+ assertEquals(DiscoverySessionStatus.INTERRUPTED, found.completionStatus)
+ }
+
+ @Test
+ fun getInterruptedSession_returnsPendingHomeRestoreSessions() = runTest {
+ createDb()
+ val address = "x:AA:BB:CC:DD:EE:FF"
+ val pendingStatuses =
+ listOf(
+ DiscoverySessionStatus.RESTORE_PENDING_COMPLETE,
+ DiscoverySessionStatus.RESTORE_PENDING_FAILED,
+ DiscoverySessionStatus.RESTORE_PENDING_STOPPED,
+ )
+
+ pendingStatuses.forEach { status ->
+ val sessionId =
+ dao.insertSession(testSession(timestamp = 1L).copy(deviceAddress = address, completionStatus = status))
+
+ val found = dao.getInterruptedSession(address)
+ assertNotNull(found, "Status $status should be treated as interrupted")
+ assertEquals(status, found.completionStatus)
+ dao.deleteSession(sessionId)
+ }
}
@Test
fun getInterruptedSession_returnsNullForDifferentDevice() = runTest {
createDb()
- dao.insertSession(testSession().copy(deviceAddress = "x:OTHER-DEVICE", completionStatus = "in_progress"))
+ dao.insertSession(
+ testSession().copy(deviceAddress = "x:OTHER-DEVICE", completionStatus = DiscoverySessionStatus.IN_PROGRESS),
+ )
assertNull(dao.getInterruptedSession("x:AA:BB:CC:DD:EE:FF"))
}
@Test
fun getInterruptedSession_returnsNullForCompletedSession() = runTest {
createDb()
- dao.insertSession(testSession().copy(deviceAddress = "x:AA:BB:CC:DD:EE:FF", completionStatus = "complete"))
+ dao.insertSession(
+ testSession()
+ .copy(deviceAddress = "x:AA:BB:CC:DD:EE:FF", completionStatus = DiscoverySessionStatus.COMPLETE),
+ )
assertNull(dao.getInterruptedSession("x:AA:BB:CC:DD:EE:FF"))
}
@@ -132,10 +245,12 @@ abstract class CommonDiscoveryDaoTest {
fun getInterruptedSession_returnsMostRecentWhenMultipleMatch() = runTest {
createDb()
dao.insertSession(
- testSession(timestamp = 1L).copy(deviceAddress = "x:AA:BB:CC:DD:EE:FF", completionStatus = "interrupted"),
+ testSession(timestamp = 1L)
+ .copy(deviceAddress = "x:AA:BB:CC:DD:EE:FF", completionStatus = DiscoverySessionStatus.INTERRUPTED),
)
dao.insertSession(
- testSession(timestamp = 2L).copy(deviceAddress = "x:AA:BB:CC:DD:EE:FF", completionStatus = "in_progress"),
+ testSession(timestamp = 2L)
+ .copy(deviceAddress = "x:AA:BB:CC:DD:EE:FF", completionStatus = DiscoverySessionStatus.IN_PROGRESS),
)
val found = dao.getInterruptedSession("x:AA:BB:CC:DD:EE:FF")
assertNotNull(found)
@@ -354,7 +469,7 @@ abstract class CommonDiscoveryDaoTest {
timestamp = timestamp,
presetsScanned = "LONG_FAST,SHORT_FAST",
homePreset = homePreset,
- completionStatus = "in_progress",
+ completionStatus = DiscoverySessionStatus.IN_PROGRESS,
)
private fun testPresetResult(sessionId: Long, presetName: String = "LONG_FAST") = DiscoveryPresetResultEntity(
diff --git a/core/database/src/commonTest/kotlin/org/meshtastic/core/database/dao/SwitchingDiscoveryDaoTest.kt b/core/database/src/commonTest/kotlin/org/meshtastic/core/database/dao/SwitchingDiscoveryDaoTest.kt
index 0fd4869756..adf49506b8 100644
--- a/core/database/src/commonTest/kotlin/org/meshtastic/core/database/dao/SwitchingDiscoveryDaoTest.kt
+++ b/core/database/src/commonTest/kotlin/org/meshtastic/core/database/dao/SwitchingDiscoveryDaoTest.kt
@@ -27,6 +27,7 @@ import kotlinx.coroutines.test.runTest
import org.meshtastic.core.database.DatabaseProvider
import org.meshtastic.core.database.MeshtasticDatabase
import org.meshtastic.core.database.entity.DiscoverySessionEntity
+import org.meshtastic.core.database.entity.DiscoverySessionStatus
import org.meshtastic.core.database.getInMemoryDatabaseBuilder
import kotlin.test.AfterTest
import kotlin.test.Test
@@ -67,6 +68,34 @@ class SwitchingDiscoveryDaoTest {
assertEquals(1, dao.getAllSessionsSnapshot().size, "reads resolve the new DB too")
}
+ @Test
+ fun statusUpdateReportsUnavailableDatabase() = runTest {
+ val sessionId =
+ dao.insertSession(session(timestamp = 1).copy(completionStatus = DiscoverySessionStatus.IN_PROGRESS))
+ provider.setWritesAvailable(false)
+
+ assertEquals(0, dao.updateSessionCompletionStatus(sessionId, DiscoverySessionStatus.FAILED))
+ assertEquals(
+ DiscoverySessionStatus.IN_PROGRESS,
+ dbA.discoveryDao().getSession(sessionId)?.completionStatus,
+ "an unavailable database must not be reported as a successful status write",
+ )
+ }
+
+ @Test
+ fun recoverableStatusUpdateReportsUnavailableDatabase() = runTest {
+ val sessionId =
+ dao.insertSession(session(timestamp = 1).copy(completionStatus = DiscoverySessionStatus.IN_PROGRESS))
+ provider.setWritesAvailable(false)
+
+ assertEquals(0, dao.updateRecoverableSessionCompletionStatus(sessionId, DiscoverySessionStatus.FAILED))
+ assertEquals(
+ DiscoverySessionStatus.IN_PROGRESS,
+ dbA.discoveryDao().getSession(sessionId)?.completionStatus,
+ "an unavailable database must not be reported as a successful recoverable status write",
+ )
+ }
+
@Test
fun flowsRelatchOntoTheCurrentDb() = runTest {
dao.insertSession(session(timestamp = 1))
@@ -93,6 +122,7 @@ class SwitchingDiscoveryDaoTest {
@OptIn(ExperimentalCoroutinesApi::class)
private class TestProvider(db: MeshtasticDatabase) : DatabaseProvider {
private val _currentDb = MutableStateFlow(db)
+ private var writesAvailable = true
override val currentDb: StateFlow<MeshtasticDatabase> = _currentDb
override fun <T> observeCurrentDb(query: (MeshtasticDatabase) -> Flow<T>): Flow<T> =
@@ -100,7 +130,12 @@ class SwitchingDiscoveryDaoTest {
override suspend fun <T> withReadDb(block: suspend (MeshtasticDatabase) -> T): T = block(_currentDb.value)
- override suspend fun <T> withDb(block: suspend (MeshtasticDatabase) -> T): T? = block(_currentDb.value)
+ override suspend fun <T> withDb(block: suspend (MeshtasticDatabase) -> T): T? =
+ if (writesAvailable) block(_currentDb.value) else null
+
+ fun setWritesAvailable(available: Boolean) {
+ writesAvailable = available
+ }
fun switchTo(db: MeshtasticDatabase) {
_currentDb.value = db
diff --git a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioController.kt b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioController.kt
index 3300a5a105..8c7dd754e8 100644
--- a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioController.kt
+++ b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/RadioController.kt
@@ -17,7 +17,9 @@
package org.meshtastic.core.repository
import kotlinx.coroutines.flow.StateFlow
+import org.meshtastic.proto.Channel
import org.meshtastic.proto.ClientNotification
+import org.meshtastic.proto.Config
/**
* Central interface for controlling the radio and mesh network.
@@ -48,6 +50,9 @@ interface RadioController :
*/
val clientNotification: StateFlow<ClientNotification?>
+ /** Monotonic transport-session generation used to reject work captured by a replaced connection. */
+ val sessionGeneration: StateFlow<Long>
+
/** Clears the current [clientNotification]. */
fun clearClientNotification()
@@ -58,6 +63,24 @@ interface RadioController :
*/
fun generatePacketId(): Int
+ /**
+ * Restores local radio settings only while [expectedDeviceAddress] is still selected.
+ *
+ * The selection check and the edit-settings transaction are serialized with [setDeviceAddress], so a delayed
+ * recovery task cannot apply one device's saved settings to a replacement radio. A concurrent device selection
+ * waits for the serialized edit-settings transaction to finish. [primaryChannel] is written before [config] when
+ * present, and both writes are idempotent for the same captured device/configuration so a failed attempt can be
+ * retried. If the channel write is accepted and the config write then fails, firmware may temporarily retain that
+ * staged channel state until the transaction is retried or the connection is replaced; callers must re-check
+ * selection ownership before every retry. Returns `false` without writing when [expectedDeviceAddress] is null or
+ * selection ownership has changed.
+ */
+ suspend fun restoreLocalConfiguration(
+ expectedDeviceAddress: String?,
+ config: Config,
+ primaryChannel: Channel? = null,
+ ): Boolean
+
/** Starts providing the phone's location to the mesh. */
fun startProvideLocation()
diff --git a/core/service/src/commonMain/kotlin/org/meshtastic/core/service/RadioControllerImpl.kt b/core/service/src/commonMain/kotlin/org/meshtastic/core/service/RadioControllerImpl.kt
index e4405b7e93..413d0f43bc 100644
--- a/core/service/src/commonMain/kotlin/org/meshtastic/core/service/RadioControllerImpl.kt
+++ b/core/service/src/commonMain/kotlin/org/meshtastic/core/service/RadioControllerImpl.kt
@@ -35,6 +35,7 @@ import org.meshtastic.core.model.ConnectionEpochs
import org.meshtastic.core.model.ConnectionLifecycle
import org.meshtastic.core.model.ConnectionState
import org.meshtastic.core.repository.AdminController
+import org.meshtastic.core.repository.AdminEditScope
import org.meshtastic.core.repository.CommandSender
import org.meshtastic.core.repository.ConnectionIdentity
import org.meshtastic.core.repository.MeshDataHandler
@@ -55,7 +56,9 @@ import org.meshtastic.core.repository.RadioInterfaceService
import org.meshtastic.core.repository.RadioSessionContext
import org.meshtastic.core.repository.ServiceRepository
import org.meshtastic.core.repository.UiPrefs
+import org.meshtastic.proto.Channel
import org.meshtastic.proto.ClientNotification
+import org.meshtastic.proto.Config
private data class AssociationSnapshot(
val identity: ConnectionIdentity?,
@@ -64,6 +67,21 @@ private data class AssociationSnapshot(
val activeSession: RadioSessionContext?,
)
+internal suspend fun restoreLocalConfigurationIfOwned(
+ expectedDeviceAddress: String?,
+ currentDeviceAddress: String?,
+ config: Config,
+ primaryChannel: Channel?,
+ editLocalSettings: suspend (suspend AdminEditScope.() -> Unit) -> Unit,
+): Boolean {
+ if (expectedDeviceAddress == null || currentDeviceAddress != expectedDeviceAddress) return false
+ editLocalSettings {
+ primaryChannel?.let { setChannel(it) }
+ setConfig(config)
+ }
+ return true
+}
+
/**
* Platform-agnostic [RadioController] composition root for any target where the service runs in-process (Desktop, iOS,
* or Android in single-process mode).
@@ -117,6 +135,8 @@ class RadioControllerImpl(
private val deviceSwitchMutex = Mutex()
+ override val sessionGeneration: StateFlow<Long> = radioInterfaceService.sessionGeneration
+
init {
// Reconcile the connection-session identity at every transport-session boundary (stop/start cycle), not only
// when the selected address changes. Without this, a same-address same-device reconnect can retain the old
@@ -223,6 +243,20 @@ class RadioControllerImpl(
override fun generatePacketId(): Int = commandSender.generatePacketId()
+ override suspend fun restoreLocalConfiguration(
+ expectedDeviceAddress: String?,
+ config: Config,
+ primaryChannel: Channel?,
+ ): Boolean = deviceSwitchMutex.withLock {
+ restoreLocalConfigurationIfOwned(
+ expectedDeviceAddress = expectedDeviceAddress,
+ currentDeviceAddress = radioInterfaceService.getDeviceAddress(),
+ config = config,
+ primaryChannel = primaryChannel,
+ editLocalSettings = { block -> editLocalSettings(block) },
+ )
+ }
+
override fun startProvideLocation() {
locationManager.restart()
}
diff --git a/core/service/src/commonTest/kotlin/org/meshtastic/core/service/RadioControllerRestoreTest.kt b/core/service/src/commonTest/kotlin/org/meshtastic/core/service/RadioControllerRestoreTest.kt
new file mode 100644
index 0000000000..c687f690b1
--- /dev/null
+++ b/core/service/src/commonTest/kotlin/org/meshtastic/core/service/RadioControllerRestoreTest.kt
@@ -0,0 +1,208 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.service
+
+import dev.mokkery.MockMode
+import dev.mokkery.answering.calls
+import dev.mokkery.answering.returns
+import dev.mokkery.every
+import dev.mokkery.everySuspend
+import dev.mokkery.matcher.any
+import dev.mokkery.mock
+import dev.mokkery.verify.VerifyMode.Companion.exactly
+import dev.mokkery.verifySuspend
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.async
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import org.meshtastic.core.common.database.DatabaseManager
+import org.meshtastic.core.repository.AdminEditScope
+import org.meshtastic.core.repository.CommandSender
+import org.meshtastic.core.repository.MeshDataHandler
+import org.meshtastic.core.repository.MeshLocationManager
+import org.meshtastic.core.repository.MeshMessageProcessor
+import org.meshtastic.core.repository.MeshPrefs
+import org.meshtastic.core.repository.NodeManager
+import org.meshtastic.core.repository.NodeRepository
+import org.meshtastic.core.repository.NotificationManager
+import org.meshtastic.core.repository.PacketRepository
+import org.meshtastic.core.repository.PlatformAnalytics
+import org.meshtastic.core.repository.RadioConfigRepository
+import org.meshtastic.core.repository.RadioInterfaceService
+import org.meshtastic.core.repository.ServiceRepository
+import org.meshtastic.core.repository.UiPrefs
+import org.meshtastic.proto.Channel
+import org.meshtastic.proto.ChannelSettings
+import org.meshtastic.proto.Config
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+class RadioControllerRestoreTest {
+ private class Fixture {
+ val commandSender: CommandSender = mock(MockMode.autofill)
+ val nodeManager: NodeManager = mock(MockMode.autofill)
+ val radioInterfaceService: RadioInterfaceService = mock(MockMode.autofill)
+ val meshPrefs: MeshPrefs = mock(MockMode.autofill)
+ val databaseManager: DatabaseManager = mock(MockMode.autofill)
+
+ private val serviceRepository: ServiceRepository = ServiceRepositoryImpl()
+ private val nodeRepository: NodeRepository = mock(MockMode.autofill)
+ private val locationManager: MeshLocationManager = mock(MockMode.autofill)
+ private val packetRepository: PacketRepository = mock(MockMode.autofill)
+ private val dataHandler: MeshDataHandler = mock(MockMode.autofill)
+ private val analytics: PlatformAnalytics = mock(MockMode.autofill)
+ private val uiPrefs: UiPrefs = mock(MockMode.autofill)
+ private val notificationManager: NotificationManager = mock(MockMode.autofill)
+ private val messageProcessor: MeshMessageProcessor = mock(MockMode.autofill)
+ private val radioConfigRepository: RadioConfigRepository = mock(MockMode.autofill)
+
+ fun create(
+ scope: CoroutineScope,
+ deviceAddress: MutableStateFlow<String?> = MutableStateFlow(null),
+ ): RadioControllerImpl {
+ every { nodeManager.myNodeNum } returns MutableStateFlow(1234)
+ every { nodeManager.myDeviceId } returns MutableStateFlow(null)
+ every { nodeManager.connectionIdentity } returns MutableStateFlow(null)
+ every { radioInterfaceService.sessionGeneration } returns MutableStateFlow(0L)
+ every { radioInterfaceService.activeSession } returns MutableStateFlow(null)
+ every { meshPrefs.deviceAddress } returns deviceAddress
+ every { radioInterfaceService.getDeviceAddress() } calls { deviceAddress.value }
+ return RadioControllerImpl(
+ serviceRepository = serviceRepository,
+ nodeRepository = nodeRepository,
+ commandSender = commandSender,
+ nodeManager = nodeManager,
+ radioInterfaceService = radioInterfaceService,
+ locationManager = locationManager,
+ packetRepository = lazy { packetRepository },
+ dataHandler = lazy { dataHandler },
+ analytics = analytics,
+ meshPrefs = meshPrefs,
+ uiPrefs = uiPrefs,
+ databaseManager = databaseManager,
+ notificationManager = notificationManager,
+ messageProcessor = lazy { messageProcessor },
+ radioConfigRepository = radioConfigRepository,
+ scope = scope,
+ )
+ }
+ }
+
+ @Test
+ fun restoreLocalConfigurationRejectsStaleDeviceOwnershipBeforeWriting() = runTest {
+ val fixture = Fixture()
+ val controller = fixture.create(backgroundScope)
+ every { fixture.radioInterfaceService.getDeviceAddress() } returns "x:CURRENT"
+
+ val restored =
+ controller.restoreLocalConfiguration(
+ expectedDeviceAddress = "x:STALE",
+ config = Config(lora = Config.LoRaConfig(use_preset = true)),
+ )
+
+ assertFalse(restored)
+ verifySuspend(exactly(0)) { fixture.commandSender.sendAdmin(any(), any(), any(), any()) }
+ }
+
+ @Test
+ fun restoreLocalConfigurationRejectsMissingDeviceOwnershipBeforeWriting() = runTest {
+ val fixture = Fixture()
+ val controller = fixture.create(backgroundScope)
+
+ val restored =
+ controller.restoreLocalConfiguration(
+ expectedDeviceAddress = null,
+ config = Config(lora = Config.LoRaConfig(use_preset = true)),
+ )
+
+ assertFalse(restored)
+ verifySuspend(exactly(0)) { fixture.commandSender.sendAdmin(any(), any(), any(), any()) }
+ }
+
+ @Test
+ fun restoreLocalConfigurationUsesOneEditTransactionInChannelThenConfigOrder() = runTest {
+ val primaryChannel = Channel(index = 0, role = Channel.Role.PRIMARY, settings = ChannelSettings())
+ val config = Config(lora = Config.LoRaConfig(use_preset = true))
+ val operations = mutableListOf<String>()
+ val editScope = mock<AdminEditScope>(MockMode.autofill)
+ everySuspend { editScope.setChannel(any()) } calls { operations += "channel" }
+ everySuspend { editScope.setConfig(any()) } calls { operations += "config" }
+ var transactions = 0
+
+ val restored =
+ restoreLocalConfigurationIfOwned(
+ expectedDeviceAddress = "x:CURRENT",
+ currentDeviceAddress = "x:CURRENT",
+ config = config,
+ primaryChannel = primaryChannel,
+ editLocalSettings = { block ->
+ transactions++
+ editScope.block()
+ },
+ )
+
+ assertTrue(restored)
+ assertEquals(1, transactions)
+ assertEquals(listOf("channel", "config"), operations)
+ verifySuspend(exactly(1)) { editScope.setChannel(primaryChannel) }
+ verifySuspend(exactly(1)) { editScope.setConfig(config) }
+ }
+
+ @Test
+ fun restoreLocalConfigurationWaitsForDeviceSelectionAndRejectsThePreviousDevice() = runTest {
+ val fixture = Fixture()
+ val selectedAddress = MutableStateFlow<String?>("x:OLD")
+ val controller = fixture.create(backgroundScope, selectedAddress)
+ val switchStarted = CompletableDeferred<Unit>()
+ val releaseSwitch = CompletableDeferred<Unit>()
+ everySuspend { fixture.databaseManager.switchActiveDatabase("x:NEW") } calls
+ {
+ switchStarted.complete(Unit)
+ releaseSwitch.await()
+ }
+ every { fixture.meshPrefs.setDeviceAddress("x:NEW") } calls { selectedAddress.value = "x:NEW" }
+ every { fixture.radioInterfaceService.setDeviceAddress("x:NEW") } calls
+ {
+ selectedAddress.value = "x:NEW"
+ true
+ }
+
+ val selection = launch { controller.setDeviceAddress("x:NEW") }
+ switchStarted.await()
+ val restore = async {
+ controller.restoreLocalConfiguration(
+ expectedDeviceAddress = "x:OLD",
+ config = Config(lora = Config.LoRaConfig(use_preset = true)),
+ )
+ }
+ runCurrent()
+
+ assertFalse(restore.isCompleted, "restore must wait behind the in-flight device selection")
+ verifySuspend(exactly(0)) { fixture.commandSender.sendAdmin(any(), any(), any(), any()) }
+
+ releaseSwitch.complete(Unit)
+ selection.join()
+
+ assertFalse(restore.await())
+ verifySuspend(exactly(0)) { fixture.commandSender.sendAdmin(any(), any(), any(), any()) }
+ }
+}
diff --git a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioController.kt b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioController.kt
index 40f4dd8207..878e737dbc 100644
--- a/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioController.kt
+++ b/core/testing/src/commonMain/kotlin/org/meshtastic/core/testing/FakeRadioController.kt
@@ -18,6 +18,8 @@ package org.meshtastic.core.testing
import kotlinx.coroutines.NonCancellable
import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
import org.meshtastic.core.model.ConnectionState
import org.meshtastic.core.model.DataPacket
import org.meshtastic.core.model.Position
@@ -164,6 +166,17 @@ class FakeRadioController :
var gattCacheInvalidationRequested = false
private set
+ private val _sessionGeneration = mutableStateFlow(0L)
+ override val sessionGeneration: StateFlow<Long> = _sessionGeneration
+
+ private val deviceSwitchMutex = Mutex()
+
+ /** Test hook that can suspend while a conditional local-configuration restore owns device selection. */
+ var beforeRestoreLocalConfiguration: suspend () -> Unit = {}
+
+ /** Device identity currently owned by the conditional restore seam. */
+ var selectedDeviceAddress: String? = null
+
/** Deterministic test hook invoked inside every [requestRebootOta] call with the request parameters. */
var onRequestRebootOta: suspend (requestId: Int, destNum: Int, mode: Int, hash: ByteArray?) -> Unit =
{ _, _, _, _ ->
@@ -203,6 +216,9 @@ class FakeRadioController :
stopProvideLocationCalled = false
onRequestRebootOta = { _, _, _, _ -> }
gattCacheInvalidationRequested = false
+ beforeRestoreLocalConfiguration = {}
+ selectedDeviceAddress = null
+ _sessionGeneration.value = 0L
}
}
@@ -255,6 +271,18 @@ class FakeRadioController :
settingsOperations.add(SettingsOperation.SetChannel(channel))
}
+ override suspend fun restoreLocalConfiguration(
+ expectedDeviceAddress: String?,
+ config: Config,
+ primaryChannel: Channel?,
+ ): Boolean = deviceSwitchMutex.withLock {
+ if (expectedDeviceAddress == null || selectedDeviceAddress != expectedDeviceAddress) return@withLock false
+ beforeRestoreLocalConfiguration()
+ primaryChannel?.let { channel -> setLocalChannel(channel) }
+ setLocalConfig(config)
+ true
+ }
+
override suspend fun setOwner(destNum: Int, user: User, packetId: Int) {
lastSetOwnerUser = user
ownerWrites.add(OwnerWrite(destination = destNum.takeUnless { it == 0 }, user = user))
@@ -413,7 +441,8 @@ class FakeRadioController :
stopProvideLocationCalled = true
}
- override suspend fun setDeviceAddress(address: String) {
+ override suspend fun setDeviceAddress(address: String) = deviceSwitchMutex.withLock {
+ selectedDeviceAddress = address
lastSetDeviceAddress = address
}
@@ -425,6 +454,10 @@ class FakeRadioController :
fun setConnectionState(state: ConnectionState) = connectionStateHolder.setConnectionState(state)
+ fun setSessionGeneration(generation: Long) {
+ _sessionGeneration.value = generation
+ }
+
fun setClientNotification(notification: ClientNotification?) {
_clientNotification.value = notification
}
diff --git a/core/testing/src/commonTest/kotlin/org/meshtastic/core/testing/FakeRadioControllerRestoreTest.kt b/core/testing/src/commonTest/kotlin/org/meshtastic/core/testing/FakeRadioControllerRestoreTest.kt
new file mode 100644
index 0000000000..93d849821a
--- /dev/null
+++ b/core/testing/src/commonTest/kotlin/org/meshtastic/core/testing/FakeRadioControllerRestoreTest.kt
@@ -0,0 +1,152 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.core.testing
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.async
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import org.meshtastic.proto.Channel
+import org.meshtastic.proto.ChannelSettings
+import org.meshtastic.proto.Config
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+import kotlin.test.assertFalse
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+
+class FakeRadioControllerRestoreTest {
+ @Test
+ fun resetClearsRestoreSessionState() = runTest {
+ val controller = FakeRadioController()
+ controller.beforeRestoreLocalConfiguration = { error("stale hook") }
+ controller.selectedDeviceAddress = "x:DEVICE"
+ controller.setSessionGeneration(7L)
+
+ controller.reset()
+
+ assertNull(controller.selectedDeviceAddress)
+ assertEquals(0L, controller.sessionGeneration.value)
+ controller.selectedDeviceAddress = "x:DEVICE"
+ assertTrue(
+ controller.restoreLocalConfiguration(
+ expectedDeviceAddress = "x:DEVICE",
+ config = Config(),
+ primaryChannel = null,
+ ),
+ )
+ }
+
+ @Test
+ fun restoreRequiresSelectedDeviceOwnership() = runTest {
+ val controller = FakeRadioController().apply { selectedDeviceAddress = "x:CURRENT" }
+ val channel = Channel(index = 0, settings = ChannelSettings(name = "Primary"))
+ val config = Config(lora = Config.LoRaConfig(use_preset = true))
+
+ assertFalse(
+ controller.restoreLocalConfiguration(
+ expectedDeviceAddress = "x:STALE",
+ config = config,
+ primaryChannel = channel,
+ ),
+ )
+ assertTrue(controller.localChannels.isEmpty())
+ assertTrue(controller.localConfigs.isEmpty())
+
+ assertTrue(
+ controller.restoreLocalConfiguration(
+ expectedDeviceAddress = "x:CURRENT",
+ config = config,
+ primaryChannel = channel,
+ ),
+ )
+ assertEquals(listOf(channel), controller.localChannels)
+ assertEquals(listOf(config), controller.localConfigs)
+ }
+
+ @Test
+ fun restoreRejectsMissingSelectedDeviceOwnership() = runTest {
+ val controller = FakeRadioController()
+ val channel = Channel(index = 0, settings = ChannelSettings(name = "Primary"))
+ val config = Config(lora = Config.LoRaConfig(use_preset = true))
+
+ val restored =
+ controller.restoreLocalConfiguration(
+ expectedDeviceAddress = null,
+ config = config,
+ primaryChannel = channel,
+ )
+
+ assertFalse(restored)
+ assertTrue(controller.localChannels.isEmpty())
+ assertTrue(controller.localConfigs.isEmpty())
+ }
+
+ @Test
+ fun restorePropagatesConfigFailureAfterChannelWrite() = runTest {
+ val controller =
+ FakeRadioController().apply {
+ selectedDeviceAddress = "x:CURRENT"
+ throwOnSetLocalConfig = true
+ }
+ val channel = Channel(index = 0, settings = ChannelSettings(name = "Primary"))
+ val config = Config(lora = Config.LoRaConfig(use_preset = true))
+
+ assertFailsWith<IllegalStateException> {
+ controller.restoreLocalConfiguration(
+ expectedDeviceAddress = "x:CURRENT",
+ config = config,
+ primaryChannel = channel,
+ )
+ }
+
+ assertEquals(listOf(channel), controller.localChannels)
+ assertTrue(controller.localConfigs.isEmpty())
+ }
+
+ @Test
+ fun restoreSerializesWithDeviceSelection() = runTest {
+ val controller = FakeRadioController()
+ controller.setDeviceAddress("x:FIRST")
+ val writeEntered = CompletableDeferred<Unit>()
+ val releaseWrite = CompletableDeferred<Unit>()
+ controller.beforeRestoreLocalConfiguration = {
+ writeEntered.complete(Unit)
+ releaseWrite.await()
+ }
+ val config = Config(lora = Config.LoRaConfig(use_preset = true))
+
+ val restore = async {
+ controller.restoreLocalConfiguration(
+ expectedDeviceAddress = "x:FIRST",
+ config = config,
+ primaryChannel = null,
+ )
+ }
+ writeEntered.await()
+ val selection = async { controller.setDeviceAddress("x:SECOND") }
+ runCurrent()
+
+ assertFalse(selection.isCompleted, "device selection must wait for the in-flight restoration")
+ releaseWrite.complete(Unit)
+ assertTrue(restore.await())
+ selection.await()
+ assertEquals("x:SECOND", controller.selectedDeviceAddress)
+ assertEquals(listOf(config), controller.localConfigs)
+ }
+}
diff --git a/feature/discovery/src/commonMain/kotlin/org/meshtastic/feature/discovery/DiscoveryHomeRestorer.kt b/feature/discovery/src/commonMain/kotlin/org/meshtastic/feature/discovery/DiscoveryHomeRestorer.kt
new file mode 100644
index 0000000000..e250ba3695
--- /dev/null
+++ b/feature/discovery/src/commonMain/kotlin/org/meshtastic/feature/discovery/DiscoveryHomeRestorer.kt
@@ -0,0 +1,473 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.feature.discovery
+
+import co.touchlab.kermit.Logger
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineStart
+import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.async
+import kotlinx.coroutines.currentCoroutineContext
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.ensureActive
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withTimeoutOrNull
+import org.meshtastic.core.common.di.ApplicationCoroutineScope
+import org.meshtastic.core.common.util.safeCatching
+import org.meshtastic.core.database.dao.DiscoveryDao
+import org.meshtastic.core.database.entity.DiscoverySessionEntity
+import org.meshtastic.core.database.entity.DiscoverySessionStatus
+import org.meshtastic.core.model.ConnectionState
+import org.meshtastic.core.repository.MeshPrefs
+import org.meshtastic.core.repository.RadioController
+import org.meshtastic.core.repository.ServiceRepository
+import org.meshtastic.proto.Channel
+import org.meshtastic.proto.ChannelSettings
+import org.meshtastic.proto.Config
+import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.seconds
+
+internal data class DiscoveryHomeRestorePlan(
+ val sessionId: Long,
+ val deviceAddress: String?,
+ val loraConfig: Config.LoRaConfig,
+ val primaryChannel: ChannelSettings?,
+ val restorePrimaryChannel: Boolean,
+ val finalStatus: String,
+)
+
+private fun DiscoveryHomeRestorePlan.matchesDevice(deviceAddress: String?): Boolean =
+ this.deviceAddress == deviceAddress
+
+internal fun finalStatusForPendingRestore(
+ completionStatus: String,
+ default: String = DiscoverySessionStatus.RESTORED,
+): String = when (completionStatus) {
+ DiscoverySessionStatus.RESTORE_PENDING_STOPPED -> DiscoverySessionStatus.STOPPED
+ DiscoverySessionStatus.RESTORE_PENDING_FAILED -> DiscoverySessionStatus.FAILED
+ DiscoverySessionStatus.RESTORE_PENDING_COMPLETE -> DiscoverySessionStatus.COMPLETE
+ else -> default
+}
+
+private suspend fun awaitRestoreResult(result: Deferred<Boolean>, timeout: kotlin.time.Duration): Boolean {
+ val completed =
+ withTimeoutOrNull(timeout) {
+ // A superseded restore cancels its Deferred while the foreground waiter remains valid. Inspect that
+ // cancellation explicitly so only cancellation of this waiting coroutine propagates.
+ val attempt = runCatching { result.await() }
+ val failure = attempt.exceptionOrNull()
+ when {
+ failure == null -> attempt.getOrDefault(false)
+
+ failure is CancellationException -> {
+ currentCoroutineContext().ensureActive()
+ false
+ }
+
+ else -> {
+ Logger.w(failure) { "DiscoveryScanEngine: awaited home restore failed" }
+ false
+ }
+ }
+ }
+ return completed == true
+}
+
+internal suspend fun awaitForegroundRestore(result: Deferred<Boolean>): Boolean =
+ awaitRestoreResult(result, DiscoveryHomeRestorer.FOREGROUND_RESTORE_TIMEOUT_MS.milliseconds)
+
+/** Owns process-lifetime restoration of the radio configuration captured before a discovery scan. */
+@Suppress("TooManyFunctions")
+internal class DiscoveryHomeRestorer(
+ private val radioController: RadioController,
+ private val serviceRepository: ServiceRepository,
+ private val discoveryDao: DiscoveryDao,
+ private val applicationScope: ApplicationCoroutineScope,
+ private val meshPrefs: MeshPrefs,
+) {
+ /** Guarded by [pendingMutex]. Retains a completed restore long enough to correct a racing terminal outcome. */
+ private class RestoreState(var finalStatus: String, var persistedFinalStatus: String? = null) {
+ val completedSuccessfully = MutableStateFlow<Boolean?>(null)
+ }
+
+ private data class RestoreRetryState(
+ var writeAttempts: Int = 0,
+ var ownershipRechecksRemaining: Int = MAX_OWNERSHIP_RECHECKS,
+ var writeRetryDelayMs: Long = RETRY_DELAY_MS,
+ )
+
+ private data class PendingRestore(
+ val plan: DiscoveryHomeRestorePlan,
+ val result: Deferred<Boolean>,
+ val state: RestoreState,
+ )
+
+ // Lock order invariant: acquire persistenceMutex before pendingMutex. Never acquire persistenceMutex while holding
+ // pendingMutex. pendingRestoreSnapshot mirrors the guarded reference for non-suspending admission rechecks only.
+ private val pendingMutex = Mutex()
+ private val persistenceMutex = Mutex()
+ private var pendingRestore: PendingRestore? = null
+ private val pendingRestoreSnapshot = MutableStateFlow<PendingRestore?>(null)
+
+ /** Caller must hold [pendingMutex]. */
+ private fun publishPendingRestore(value: PendingRestore?) {
+ pendingRestore = value
+ pendingRestoreSnapshot.value = value
+ }
+
+ /** A same-device scan cannot retune until a previously scheduled home restore has succeeded. */
+ suspend fun awaitBeforeScan(deviceAddress: String?): Boolean {
+ val pending = pendingMutex.withLock { pendingRestore }
+ return when {
+ pending == null -> true
+
+ pending.plan.deviceAddress != deviceAddress -> {
+ pending.result.cancel()
+ pendingMutex.withLock { if (pendingRestore === pending) publishPendingRestore(null) }
+ true
+ }
+
+ else -> {
+ if (!pending.result.isCompleted) {
+ Logger.i { "DiscoveryScanEngine: waiting for pending home restore before starting a new scan" }
+ }
+ val restored = awaitRestoreResult(pending.result, START_WAIT_TIMEOUT)
+ if (!restored) {
+ if (pending.result.isCompleted) {
+ Logger.e {
+ "DiscoveryScanEngine: previous home restore did not complete successfully; " +
+ "refusing a same-device scan"
+ }
+ } else {
+ Logger.w { "DiscoveryScanEngine: home restore is still pending; deferring new scan" }
+ }
+ }
+ restored
+ }
+ }
+ }
+
+ /**
+ * Non-suspending scan-engine recheck used while its mutex prevents a new recovery restore from racing admission. A
+ * completed restore remains blocking unless it completed successfully.
+ */
+ fun hasUnsatisfiedRestoreFor(deviceAddress: String?): Boolean {
+ val pending = pendingRestoreSnapshot.value?.takeIf { it.plan.deviceAddress == deviceAddress }
+ return pending?.let { !it.result.isCompleted || it.state.completedSuccessfully.value != true } ?: false
+ }
+
+ /** Registers a restore in the application scope. Repeated scheduling of the same active plan is idempotent. */
+ suspend fun schedule(plan: DiscoveryHomeRestorePlan): Deferred<Boolean> {
+ var superseded: PendingRestore? = null
+ var created = false
+ val pending =
+ pendingMutex.withLock {
+ val existing = pendingRestore
+ if (
+ existing != null &&
+ !existing.result.isCompleted &&
+ existing.plan.sessionId == plan.sessionId &&
+ existing.plan.deviceAddress == plan.deviceAddress
+ ) {
+ existing
+ } else {
+ superseded = existing?.takeUnless { it.result.isCompleted }
+ val state = RestoreState(plan.finalStatus)
+ val result =
+ applicationScope.async(start = CoroutineStart.LAZY) {
+ restoreUntilComplete(plan, state).also { state.completedSuccessfully.value = it }
+ }
+ PendingRestore(plan, result, state).also {
+ publishPendingRestore(it)
+ created = true
+ }
+ }
+ }
+ superseded?.result?.cancel()
+ if (created) {
+ pending.result.invokeOnCompletion { cause ->
+ if (cause != null && cause !is CancellationException) {
+ Logger.e(cause) { "DiscoveryScanEngine: background home restore failed unexpectedly" }
+ }
+ }
+ pending.result.start()
+ }
+ return pending.result
+ }
+
+ /** Changes the status a running or just-completed restore publishes, correcting a racing terminal write. */
+ suspend fun updateFinalStatus(sessionId: Long, finalStatus: String) {
+ val pending =
+ pendingMutex.withLock {
+ val current = pendingRestore?.takeIf { it.plan.sessionId == sessionId } ?: return@withLock null
+ current.state.finalStatus = finalStatus
+ current
+ } ?: return
+ val correction =
+ persistenceMutex.withLock {
+ val shouldCorrect =
+ pendingMutex.withLock {
+ pendingRestore === pending &&
+ pending.state.finalStatus == finalStatus &&
+ pending.state.persistedFinalStatus?.let { it != finalStatus } == true
+ }
+ if (!shouldCorrect) return@withLock null
+
+ val result = safeCatching { discoveryDao.updateSessionCompletionStatus(sessionId, finalStatus) }
+ if (result.getOrNull() == 1) {
+ pendingMutex.withLock {
+ if (pendingRestore === pending) pending.state.persistedFinalStatus = finalStatus
+ }
+ }
+ result
+ }
+ val correctionFailure = correction?.exceptionOrNull()
+ when {
+ correctionFailure != null ->
+ Logger.e(correctionFailure) {
+ "DiscoveryScanEngine: failed to correct the restored session's terminal status"
+ }
+
+ correction != null && correction.getOrNull() != 1 ->
+ Logger.w { "DiscoveryScanEngine: terminal status correction did not update session $sessionId" }
+ }
+ }
+
+ /** Gives normal scan completion a bounded foreground opportunity while the process-lifetime job keeps running. */
+ suspend fun awaitForeground(plan: DiscoveryHomeRestorePlan): Boolean =
+ awaitRestoreResult(schedule(plan), FOREGROUND_RESTORE_TIMEOUT_MS.milliseconds)
+
+ /** Registers a persisted interrupted/pending session restore without waiting for it to finish. */
+ suspend fun schedulePersistedSession(session: DiscoverySessionEntity): Deferred<Boolean>? {
+ val loraConfig = session.homeLoraConfig ?: return null
+ return schedule(
+ DiscoveryHomeRestorePlan(
+ sessionId = session.id,
+ deviceAddress = session.deviceAddress,
+ loraConfig = loraConfig,
+ primaryChannel = session.homePrimaryChannel,
+ restorePrimaryChannel = session.homePrimaryChannel != null,
+ finalStatus = finalStatusForPendingRestore(session.completionStatus),
+ ),
+ )
+ }
+
+ private suspend fun restoreUntilComplete(plan: DiscoveryHomeRestorePlan, state: RestoreState): Boolean {
+ if (plan.restorePrimaryChannel && plan.primaryChannel == null) {
+ Logger.e {
+ "DiscoveryScanEngine: primary-channel restore required but no channel captured " +
+ "for session ${plan.sessionId}; abandoning restore"
+ }
+ markUnrestorableAndReleaseBarrier(plan)
+ return false
+ }
+ val retries = RestoreRetryState()
+ var restored = false
+ var shouldContinue = true
+ while (
+ currentCoroutineContext().isActive &&
+ plan.matchesDevice(meshPrefs.deviceAddress.value) &&
+ !restored &&
+ shouldContinue
+ ) {
+ if (awaitConnected(plan)) {
+ val attempt = safeCatching { applyHomeConfiguration(plan) }
+ val failure = attempt.exceptionOrNull()
+ restored = attempt.getOrDefault(false)
+ shouldContinue =
+ when {
+ restored -> {
+ finalizeRecoveredSessionBestEffort(plan.sessionId, state)
+ false
+ }
+
+ failure == null -> retryAfterOwnershipRejection(plan, retries)
+
+ else -> retryAfterWriteFailure(plan, retries, failure)
+ }
+ }
+ }
+ return restored
+ }
+
+ /** Gives transport and selected-device ownership a short window to converge without retrying forever. */
+ private suspend fun retryAfterOwnershipRejection(
+ plan: DiscoveryHomeRestorePlan,
+ retries: RestoreRetryState,
+ ): Boolean {
+ if (retries.ownershipRechecksRemaining == MAX_OWNERSHIP_RECHECKS) {
+ Logger.i {
+ "DiscoveryScanEngine: home restore rejected by device ownership for session " +
+ "${plan.sessionId}; re-checking selection"
+ }
+ }
+ val shouldRetry = retries.ownershipRechecksRemaining > 0
+ if (shouldRetry) {
+ retries.ownershipRechecksRemaining--
+ delay(RETRY_DELAY_MS)
+ } else {
+ Logger.w {
+ "DiscoveryScanEngine: home restore still lacks device ownership for session " +
+ "${plan.sessionId}; leaving it recoverable for a later connection change"
+ }
+ }
+ return shouldRetry
+ }
+
+ /** Retries real radio writes only while this restore still owns the connected device. */
+ private suspend fun retryAfterWriteFailure(
+ plan: DiscoveryHomeRestorePlan,
+ retries: RestoreRetryState,
+ failure: Throwable,
+ ): Boolean {
+ val stillOwnsConnectedDevice =
+ plan.matchesDevice(meshPrefs.deviceAddress.value) &&
+ serviceRepository.connectionState.value is ConnectionState.Connected
+ if (!stillOwnsConnectedDevice) return true
+
+ retries.ownershipRechecksRemaining = MAX_OWNERSHIP_RECHECKS
+ retries.writeAttempts++
+ val attemptsExhausted = retries.writeAttempts >= MAX_RESTORE_ATTEMPTS
+ Logger.w(failure) {
+ if (attemptsExhausted) {
+ "DiscoveryScanEngine: final home restore attempt failed"
+ } else {
+ "DiscoveryScanEngine: home restore attempt failed; retrying when available"
+ }
+ }
+ return if (attemptsExhausted) {
+ Logger.e {
+ "DiscoveryScanEngine: home restore exhausted $MAX_RESTORE_ATTEMPTS attempts for session " +
+ "${plan.sessionId}; marking it unrestorable"
+ }
+ markUnrestorableAndReleaseBarrier(plan)
+ false
+ } else {
+ delay(retries.writeRetryDelayMs)
+ retries.writeRetryDelayMs =
+ (retries.writeRetryDelayMs * RETRY_BACKOFF_MULTIPLIER).coerceAtMost(MAX_RETRY_DELAY_MS)
+ true
+ }
+ }
+
+ private suspend fun awaitConnected(plan: DiscoveryHomeRestorePlan): Boolean {
+ if (!plan.matchesDevice(meshPrefs.deviceAddress.value)) return false
+ if (serviceRepository.connectionState.value !is ConnectionState.Connected) {
+ combine(serviceRepository.connectionState, meshPrefs.deviceAddress) { state, address ->
+ state is ConnectionState.Connected || !plan.matchesDevice(address)
+ }
+ .first { it }
+ }
+ return plan.matchesDevice(meshPrefs.deviceAddress.value) &&
+ serviceRepository.connectionState.value is ConnectionState.Connected
+ }
+
+ private suspend fun applyHomeConfiguration(plan: DiscoveryHomeRestorePlan): Boolean {
+ val primaryChannel =
+ plan.primaryChannel
+ ?.takeIf { plan.restorePrimaryChannel }
+ ?.let { Channel(index = 0, role = Channel.Role.PRIMARY, settings = it) }
+ val restored =
+ radioController.restoreLocalConfiguration(
+ expectedDeviceAddress = plan.deviceAddress,
+ config = Config(lora = plan.loraConfig),
+ primaryChannel = primaryChannel,
+ )
+ if (!restored) return false
+ Logger.i { "DiscoveryScanEngine: restored original LoRa config for session ${plan.sessionId}" }
+ delay(POST_RESTORE_SETTLE_DELAY_MS)
+ return true
+ }
+
+ private suspend fun finalizeRecoveredSessionBestEffort(sessionId: Long, state: RestoreState) {
+ val result =
+ persistenceMutex.withLock {
+ val finalStatus = pendingMutex.withLock { state.finalStatus }
+ val persistence = safeCatching {
+ discoveryDao.updateRecoverableSessionCompletionStatus(sessionId, finalStatus)
+ }
+ if (persistence.getOrNull() == 1) {
+ pendingMutex.withLock { state.persistedFinalStatus = finalStatus }
+ }
+ persistence
+ }
+ val failure = result.exceptionOrNull()
+ if (failure != null) {
+ Logger.e(failure) {
+ "DiscoveryScanEngine: radio restored but terminal session persistence failed; keeping recovery row"
+ }
+ } else if (result.getOrNull() != 1) {
+ Logger.w {
+ "DiscoveryScanEngine: session $sessionId was no longer recoverable; terminal status not written"
+ }
+ }
+ }
+
+ /**
+ * Makes an unrecoverable restore terminal in both durable and in-memory ownership. The admission barrier is
+ * released only after the terminal status is persisted; if persistence fails, the recoverable row and barrier
+ * remain available for a later recovery attempt.
+ */
+ private suspend fun markUnrestorableAndReleaseBarrier(plan: DiscoveryHomeRestorePlan) {
+ if (!markUnrestorableBestEffort(plan.sessionId)) return
+ pendingMutex.withLock { if (pendingRestore?.plan == plan) publishPendingRestore(null) }
+ }
+
+ private suspend fun markUnrestorableBestEffort(sessionId: Long): Boolean {
+ val result =
+ persistenceMutex.withLock {
+ safeCatching {
+ discoveryDao.updateRecoverableSessionCompletionStatus(
+ sessionId,
+ DiscoverySessionStatus.UNRESTORABLE,
+ )
+ }
+ }
+ val failure = result.exceptionOrNull()
+ return when {
+ failure != null -> {
+ Logger.e(failure) { "DiscoveryScanEngine: failed to persist unrestorable session status" }
+ false
+ }
+
+ result.getOrNull() != 1 -> {
+ Logger.w {
+ "DiscoveryScanEngine: session $sessionId was no longer recoverable; unrestorable not written"
+ }
+ false
+ }
+
+ else -> true
+ }
+ }
+
+ internal companion object {
+ const val FOREGROUND_RESTORE_TIMEOUT_MS = 90_000L
+ const val RETRY_DELAY_MS = 1_000L
+ internal const val MAX_RETRY_DELAY_MS = 30_000L
+ internal const val MAX_RESTORE_ATTEMPTS = 7
+ internal const val MAX_OWNERSHIP_RECHECKS = 15
+ internal const val RETRY_BACKOFF_MULTIPLIER = 2L
+ const val POST_RESTORE_SETTLE_DELAY_MS = 3_000L
+ private val START_WAIT_TIMEOUT = 15.seconds
+ }
+}
diff --git a/feature/discovery/src/commonMain/kotlin/org/meshtastic/feature/discovery/DiscoveryInterruptedSessionRecovery.kt b/feature/discovery/src/commonMain/kotlin/org/meshtastic/feature/discovery/DiscoveryInterruptedSessionRecovery.kt
new file mode 100644
index 0000000000..92f1597171
--- /dev/null
+++ b/feature/discovery/src/commonMain/kotlin/org/meshtastic/feature/discovery/DiscoveryInterruptedSessionRecovery.kt
@@ -0,0 +1,111 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.feature.discovery
+
+import co.touchlab.kermit.Logger
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.currentCoroutineContext
+import kotlinx.coroutines.ensureActive
+import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.launch
+import org.meshtastic.core.common.util.safeCatching
+import org.meshtastic.core.database.dao.DiscoveryDao
+import org.meshtastic.core.database.entity.DiscoverySessionEntity
+import org.meshtastic.core.database.entity.DiscoverySessionStatus
+import org.meshtastic.core.model.ConnectionState
+import org.meshtastic.core.repository.MeshPrefs
+import org.meshtastic.core.repository.ServiceRepository
+
+/** Restores persisted discovery sessions only while their original radio still owns the recovery. */
+internal class DiscoveryInterruptedSessionRecovery(
+ private val serviceRepository: ServiceRepository,
+ private val discoveryDao: DiscoveryDao,
+ private val meshPrefs: MeshPrefs,
+ private val isScanActive: suspend () -> Boolean,
+ private val scheduleRestoreIfIdle: suspend (DiscoverySessionEntity) -> Deferred<Boolean>?,
+) {
+ suspend fun watch(onRestored: suspend (homePreset: String) -> Unit = {}): Unit = coroutineScope {
+ var recoveryJob: Job? = null
+ var recoveryDeviceAddress: String? = null
+ combine(serviceRepository.connectionState, meshPrefs.deviceAddress) { state, address -> state to address }
+ .collect { (state, address) ->
+ if (state !is ConnectionState.Connected || address == null) return@collect
+ if (recoveryJob?.isActive == true && recoveryDeviceAddress == address) return@collect
+
+ // A restore can wait through a long disconnect. Device identity participates in ownership so a switch
+ // while Connected cancels the old waiter and immediately evaluates the replacement radio.
+ recoveryJob?.cancel()
+ recoveryDeviceAddress = address
+ recoveryJob = launch {
+ val result = safeCatching { restoreIfAny(onRestored) }
+ result.exceptionOrNull()?.let { failure ->
+ Logger.w(failure) { "DiscoveryScanEngine: interrupted-session restore failed; will retry" }
+ }
+ }
+ }
+ }
+
+ private suspend fun restoreIfAny(onRestored: suspend (homePreset: String) -> Unit) {
+ val address = meshPrefs.deviceAddress.value
+ val session =
+ if (address != null && !isScanActive()) {
+ discoveryDao.getInterruptedSession(address)
+ } else {
+ null
+ }
+ val recoverable = session?.takeIf { !isScanActive() && meshPrefs.deviceAddress.value == address }
+
+ when {
+ recoverable == null -> return
+
+ recoverable.homeLoraConfig == null -> {
+ val finalStatus =
+ finalStatusForPendingRestore(
+ recoverable.completionStatus,
+ default = DiscoverySessionStatus.UNRESTORABLE,
+ )
+ val updatedRows = discoveryDao.updateRecoverableSessionCompletionStatus(recoverable.id, finalStatus)
+ if (updatedRows != 1) {
+ Logger.w {
+ "DiscoveryScanEngine: session ${recoverable.id} was no longer recoverable; " +
+ "$finalStatus not written"
+ }
+ }
+ }
+
+ else -> {
+ Logger.w { "DiscoveryScanEngine: restoring home config after interrupted session ${recoverable.id}" }
+ val restore = scheduleRestoreIfIdle(recoverable) ?: return
+ if (awaitForegroundRestore(restore)) {
+ onRestored(recoverable.homePreset)
+ } else {
+ // Foreground waiting is bounded, but the application-scope restore can continue. Await the same
+ // Deferred here so a success racing the timeout, or completing later, is still surfaced
+ // exactly once.
+ val lateResult = runCatching { restore.await() }
+ if (lateResult.exceptionOrNull() is CancellationException) {
+ currentCoroutineContext().ensureActive()
+ }
+ if (lateResult.getOrDefault(false)) onRestored(recoverable.homePreset)
+ }
+ }
+ }
+ }
+}
diff --git a/feature/discovery/src/commonMain/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngine.kt b/feature/discovery/src/commonMain/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngine.kt
index 5b7fee1153..94803efa58 100644
--- a/feature/discovery/src/commonMain/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngine.kt
+++ b/feature/discovery/src/commonMain/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngine.kt
@@ -24,7 +24,9 @@ import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
+import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.delay
+import kotlinx.coroutines.ensureActive
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
@@ -41,6 +43,7 @@ import org.meshtastic.core.database.dao.DiscoveryDao
import org.meshtastic.core.database.entity.DiscoveredNodeEntity
import org.meshtastic.core.database.entity.DiscoveryPresetResultEntity
import org.meshtastic.core.database.entity.DiscoverySessionEntity
+import org.meshtastic.core.database.entity.DiscoverySessionStatus
import org.meshtastic.core.di.CoroutineDispatchers
import org.meshtastic.core.model.ChannelOption
import org.meshtastic.core.model.ConnectionState
@@ -71,7 +74,7 @@ import org.meshtastic.proto.Telemetry
* persists aggregated results via [DiscoveryDao].
*/
@Single
-@Suppress("LongParameterList")
+@Suppress("LargeClass", "LongParameterList")
class DiscoveryScanEngine(
private val radioController: RadioController,
private val serviceRepository: ServiceRepository,
@@ -104,6 +107,33 @@ class DiscoveryScanEngine(
// region Internal scan state
private val mutex = Mutex()
+ private val homeRestorer =
+ DiscoveryHomeRestorer(radioController, serviceRepository, discoveryDao, applicationScope, meshPrefs)
+ private val terminalCoordinator =
+ DiscoveryTerminalCoordinator(
+ discoveryDao = discoveryDao,
+ homeRestorer = homeRestorer,
+ applicationScope = applicationScope,
+ onSessionUpdated = { updated -> _currentSession.value = updated },
+ onTerminalCompleted = { outcome -> _scanState.value = DiscoveryScanState.Complete(outcome) },
+ cancelScan = { mutex.withLock { cancelScanInternal() } },
+ )
+ private val interruptedSessionRecovery =
+ DiscoveryInterruptedSessionRecovery(
+ serviceRepository = serviceRepository,
+ discoveryDao = discoveryDao,
+ meshPrefs = meshPrefs,
+ isScanActive = { mutex.withLock { isActive } },
+ scheduleRestoreIfIdle = { session ->
+ mutex.withLock {
+ if (isActive || meshPrefs.deviceAddress.value != session.deviceAddress) {
+ null
+ } else {
+ homeRestorer.schedulePersistedSession(session)
+ }
+ }
+ },
+ )
private var scanScope: CoroutineScope? = null
private var dwellJob: Job? = null
private var originalLoRaConfig: Config.LoRaConfig? = null
@@ -123,6 +153,9 @@ class DiscoveryScanEngine(
private var currentPresetName: String = ""
private var totalDwellSeconds: Long = 0
+
+ /** Guards exact-once result insertion when terminal cleanup races normal dwell completion. Protected by [mutex]. */
+ private var currentDwellPersisted: Boolean = false
private var lastLocalStats: org.meshtastic.proto.LocalStats? = null
// endregion
@@ -166,94 +199,192 @@ class DiscoveryScanEngine(
require(targets.isNotEmpty()) { "At least one scan target is required" }
require(dwellDurationSeconds > 0) { "Dwell duration must be positive" }
+ if (!prepareScanStart(meshPrefs.deviceAddress.value)) return
+
+ // These repository flows may suspend while their initial snapshots load. Keep that wait outside the engine
+ // mutex so stop/reset and packet collection remain responsive, then revalidate device and restore ownership
+ // under the mutex before publishing any scan state. Capture the baseline after admission waits so a transport
+ // restart during those waits is not misreported as a change that happened during snapshot preparation.
+ val deviceAddress = meshPrefs.deviceAddress.value
+ val sessionGeneration = radioController.sessionGeneration.value
+ val snapshot = readInitialConfigSnapshot()
+ if (snapshot == null) {
+ refuseScan("Radio configuration snapshot is not available")
+ return
+ }
+ val (homeLora, initialPrimaryChannel) = snapshot
mutex.withLock {
if (isActive) {
Logger.w { "DiscoveryScanEngine: scan already active, ignoring startScan" }
- return
+ } else if (!isScanPreparationCurrent(deviceAddress, sessionGeneration)) {
+ _scanState.value = DiscoveryScanState.Failed("Selected radio changed while preparing the scan")
+ } else if (homeRestorer.hasUnsatisfiedRestoreFor(meshPrefs.deviceAddress.value)) {
+ // Interrupted-session recovery registers its restore under this mutex. Re-check here so a restore that
+ // won the race after the first wait cannot retune the radio underneath this scan.
+ _scanState.value = DiscoveryScanState.Failed("Home configuration restoration is incomplete")
+ } else {
+ _scanState.value = DiscoveryScanState.Preparing
+ val requiresPrimaryRestore = targets.any { it.channel != null }
+ when {
+ homeLora == null -> refuseUnrestorableScanLocked("Home LoRa configuration is not available")
+
+ initialPrimaryChannel == null && requiresPrimaryRestore ->
+ refuseUnrestorableScanLocked("Primary channel is not available for a custom-channel scan")
+
+ else ->
+ prepareScanLocked(
+ targets = targets,
+ dwellDurationSeconds = dwellDurationSeconds,
+ initialLoraConfig = homeLora,
+ initialPrimaryChannel = initialPrimaryChannel,
+ requiresPrimaryRestore = requiresPrimaryRestore,
+ deviceAddress = deviceAddress,
+ sessionGeneration = sessionGeneration,
+ )
+ }
}
+ }
+ }
- _scanState.value = DiscoveryScanState.Preparing
-
- // Capture the entire original LoRa config and primary channel to restore them accurately later.
- val initialLoraConfig = radioConfigRepository.localConfigFlow.first().lora
- originalLoRaConfig = initialLoraConfig
- originalPrimaryChannel = radioConfigRepository.channelSetFlow.first().settings.firstOrNull()
- tunedPrimaryChannel = false
+ private suspend fun prepareScanStart(deviceAddress: String?): Boolean = when {
+ deviceAddress == null -> {
+ refuseScan("Selected radio is not available")
+ false
+ }
- // A custom-channel target overwrites the primary channel; without a captured original we could not restore
- // it and would strand the radio on the beacon's channel. Abort rather than proceed silently.
- if (originalPrimaryChannel == null && targets.any { it.channel != null }) {
- Logger.w { "DiscoveryScanEngine: primary channel not captured; aborting custom-channel scan" }
- _scanState.value = DiscoveryScanState.Idle
- return
- }
+ !terminalCoordinator.resetForScan() -> {
+ refuseScan("Previous discovery scan cleanup is still running")
+ false
+ }
- val homePresetStr =
- if (initialLoraConfig?.use_preset == true) {
- ChannelOption.from(initialLoraConfig.modem_preset)?.name ?: ChannelOption.DEFAULT.name
- } else {
- "CUSTOM"
- }
+ !homeRestorer.awaitBeforeScan(deviceAddress) -> {
+ refuseScan("Home configuration restoration is incomplete")
+ false
+ }
- val myNodeNum = nodeRepository.myNodeInfo.value?.myNodeNum
- val myPosition = myNodeNum?.let { nodeRepository.nodeDBbyNum.value[it]?.position }
- val latDouble = (myPosition?.latitude_i ?: 0).toDouble() / POSITION_DIVISOR
- val lonDouble = (myPosition?.longitude_i ?: 0).toDouble() / POSITION_DIVISOR
+ else -> true
+ }
- // Create the DB session. homeLoraConfig/homePrimaryChannel/deviceAddress let a later reconnect detect and
- // restore this exact config if the process dies mid-scan before restoreHomePreset() ever runs.
- val session =
- DiscoverySessionEntity(
- timestamp = nowMillis,
- presetsScanned = targets.joinToString(",") { it.label },
- homePreset = homePresetStr,
- completionStatus = "in_progress",
- userLatitude = latDouble,
- userLongitude = lonDouble,
- deviceAddress = meshPrefs.deviceAddress.value,
- homeLoraConfig = initialLoraConfig,
- // Only custom-channel targets ever overwrite the primary channel, so only they need it restored —
- // mirrors the tunedPrimaryChannel gating in restoreHomePreset().
- homePrimaryChannel = originalPrimaryChannel.takeIf { targets.any { t -> t.channel != null } },
- )
- sessionId = discoveryDao.insertSession(session)
- _currentSession.value = session.copy(id = sessionId)
+ private suspend fun readInitialConfigSnapshot(): Pair<Config.LoRaConfig?, ChannelSettings?>? =
+ withTimeoutOrNull(CONFIG_SNAPSHOT_TIMEOUT_MS) {
+ radioConfigRepository.localConfigFlow.first().lora to
+ radioConfigRepository.channelSetFlow.first().settings.firstOrNull()
+ }
- // Register as packet collector
- collectorRegistry.collector = this
+ private suspend fun refuseScan(reason: String) {
+ mutex.withLock { if (!isActive) _scanState.value = DiscoveryScanState.Failed(reason) }
+ }
- // Set initial state so the scan loop's isActive guard succeeds
- _scanState.value = DiscoveryScanState.Shifting(targets.first().label)
- currentPresetName = targets.first().label
- totalDwellSeconds = dwellDurationSeconds
+ private fun refuseUnrestorableScanLocked(reason: String) {
+ Logger.w { "DiscoveryScanEngine: refusing unrestorable scan: $reason" }
+ _scanState.value = DiscoveryScanState.Failed(reason)
+ }
- // Launch scan coroutine
- val scope = CoroutineScope(dispatchers.io + SupervisorJob())
+ private suspend fun prepareScanLocked(
+ targets: List<ScanTarget>,
+ dwellDurationSeconds: Long,
+ initialLoraConfig: Config.LoRaConfig,
+ initialPrimaryChannel: ChannelSettings?,
+ requiresPrimaryRestore: Boolean,
+ deviceAddress: String?,
+ sessionGeneration: Long,
+ ) {
+ originalLoRaConfig = initialLoraConfig
+ originalPrimaryChannel = initialPrimaryChannel
+ tunedPrimaryChannel = false
+ val homePresetStr =
+ if (initialLoraConfig.use_preset) {
+ ChannelOption.from(initialLoraConfig.modem_preset)?.name ?: ChannelOption.DEFAULT.name
+ } else {
+ "CUSTOM"
+ }
+ val myNodeNum = nodeRepository.myNodeInfo.value?.myNodeNum
+ val myPosition = myNodeNum?.let { nodeRepository.nodeDBbyNum.value[it]?.position }
+ val session =
+ DiscoverySessionEntity(
+ timestamp = nowMillis,
+ presetsScanned = targets.joinToString(",") { it.label },
+ homePreset = homePresetStr,
+ completionStatus = DiscoverySessionStatus.IN_PROGRESS,
+ userLatitude = (myPosition?.latitude_i ?: 0).toDouble() / POSITION_DIVISOR,
+ userLongitude = (myPosition?.longitude_i ?: 0).toDouble() / POSITION_DIVISOR,
+ deviceAddress = deviceAddress,
+ homeLoraConfig = initialLoraConfig,
+ homePrimaryChannel = initialPrimaryChannel.takeIf { requiresPrimaryRestore },
+ )
+ val insertedSessionId = discoveryDao.insertSession(session)
+ if (!isScanPreparationCurrent(deviceAddress, sessionGeneration)) {
+ discoveryDao.deleteSession(insertedSessionId)
+ sessionId = 0L
+ originalLoRaConfig = null
+ originalPrimaryChannel = null
+ _scanState.value = DiscoveryScanState.Failed("Selected radio changed while preparing the scan")
+ return
+ }
+ sessionId = insertedSessionId
+ _currentSession.value = session.copy(id = sessionId)
+ collectorRegistry.collector = this
+ _scanState.value = DiscoveryScanState.Shifting(targets.first().label)
+ currentPresetName = targets.first().label
+ totalDwellSeconds = dwellDurationSeconds
+ currentDwellPersisted = false
+ CoroutineScope(dispatchers.io + SupervisorJob()).also { scope ->
scanScope = scope
scope.launch { runScanLoop(targets, dwellDurationSeconds) }
}
}
+ private fun isScanPreparationCurrent(deviceAddress: String?, sessionGeneration: Long): Boolean =
+ meshPrefs.deviceAddress.value == deviceAddress &&
+ radioController.sessionGeneration.value == sessionGeneration &&
+ serviceRepository.connectionState.value is ConnectionState.Connected
+
/** Stops the active scan and restores the home preset. */
suspend fun stopScan() {
- mutex.withLock {
- if (!isActive) return
- Logger.i { "DiscoveryScanEngine: stopping scan" }
- _scanState.value = DiscoveryScanState.Cancelling
- cancelScanInternal()
+ val request =
+ mutex.withLock {
+ if (!isActive) {
+ null
+ } else if (_scanState.value is DiscoveryScanState.Analysis) {
+ Logger.i { "DiscoveryScanEngine: ignoring stop after terminal analysis has started" }
+ null
+ } else {
+ if (_scanState.value !is DiscoveryScanState.Cancelling) {
+ Logger.i { "DiscoveryScanEngine: stopping scan" }
+ _scanState.value = DiscoveryScanState.Cancelling
+ }
+ // Freeze the scan generation before snapshotting its restore plan.
+ // No later target shift may race this terminal request.
+ cancelScanInternal()
+ terminalRequestLocked(
+ pendingStatus = DiscoverySessionStatus.RESTORE_PENDING_STOPPED,
+ outcome = DiscoveryScanState.CompletionOutcome.Cancelled,
+ awaitRestore = false,
+ generateAi = false,
+ )
+ }
+ }
+ if (request != null) {
+ terminalCoordinator.complete(request = request, beforeFinalize = ::persistCurrentDwellResults)
}
- persistCurrentDwellResults()
- finalizeSession("stopped")
- _scanState.value = DiscoveryScanState.Complete(DiscoveryScanState.CompletionOutcome.Cancelled)
-
- // Restore home preset in the background so we don't block the UI with the connection wait
- applicationScope.launch { restoreHomePreset() }
}
/** Resets engine state after the UI has acknowledged completion. */
fun reset() {
- _scanState.value = DiscoveryScanState.Idle
- _currentSession.value = null
+ if (!mutex.tryLock()) {
+ Logger.w { "DiscoveryScanEngine: ignoring reset while scan state is changing" }
+ return
+ }
+ try {
+ if (isActive) {
+ Logger.w { "DiscoveryScanEngine: ignoring reset while scan cleanup is still active" }
+ return
+ }
+ _currentSession.value = null
+ _scanState.value = DiscoveryScanState.Idle
+ } finally {
+ mutex.unlock()
+ }
}
// endregion
@@ -311,22 +442,38 @@ class DiscoveryScanEngine(
for (target in targets) {
if (!isActive) return
- currentPresetName = target.label
- mutex.withLock {
- collectedNodes.clear()
- deviceMetricsLog.clear()
- lastLocalStats = null
- }
- totalDwellSeconds = dwellDurationSeconds
+ val shouldShift =
+ mutex.withLock {
+ if (!canAdvanceScanLocked()) {
+ false
+ } else {
+ currentPresetName = target.label
+ totalDwellSeconds = dwellDurationSeconds
+ currentDwellPersisted = false
+ collectedNodes.clear()
+ deviceMetricsLog.clear()
+ lastLocalStats = null
+ _scanState.value = DiscoveryScanState.Shifting(target.label)
+ true
+ }
+ }
+ if (!shouldShift) return
// Shift to the new target (preset, plus a custom primary channel for beacon-channel targets)
- _scanState.value = DiscoveryScanState.Shifting(target.label)
- shiftTarget(target)
+ try {
+ shiftTarget(target)
+ } catch (e: CancellationException) {
+ throw e
+ } catch (@Suppress("TooGenericExceptionCaught") e: Exception) {
+ Logger.e(e) { "DiscoveryScanEngine: target shift failed; aborting scan and restoring home config" }
+ pauseAndAbort()
+ return
+ }
// Wait for reconnection
- _scanState.value = DiscoveryScanState.Reconnecting(target.label)
+ if (!transitionScanState(DiscoveryScanState.Reconnecting(target.label))) return
if (!waitForConnection()) {
- pauseAndAbort()
+ pauseAndAbort(persistPartialDwell = true)
return
}
@@ -335,7 +482,7 @@ class DiscoveryScanEngine(
// Dwell
if (!runDwell(target.label, dwellDurationSeconds)) {
- pauseAndAbort()
+ pauseAndAbort(persistPartialDwell = true)
return
}
if (!isActive) return
@@ -344,28 +491,45 @@ class DiscoveryScanEngine(
persistCurrentDwellResults()
}
- // All presets scanned — unregister packet collector before analysis
- collectorRegistry.collector = null
- _scanState.value = DiscoveryScanState.Analysis
- restoreHomePreset()
- generateAiSummaries()
- finalizeSession("complete")
- _scanState.value = DiscoveryScanState.Complete(DiscoveryScanState.CompletionOutcome.Success)
+ // Elect normal completion under the same mutex used by stopScan so a late stop cannot replace its outcome.
+ val request =
+ mutex.withLock {
+ collectorRegistry.collector = null
+ _scanState.value = DiscoveryScanState.Analysis
+ terminalRequestLocked(
+ pendingStatus = DiscoverySessionStatus.RESTORE_PENDING_COMPLETE,
+ outcome = DiscoveryScanState.CompletionOutcome.Success,
+ awaitRestore = true,
+ generateAi = true,
+ )
+ }
+ // complete() cancels scanScope before terminal cleanup finishes. This scan coroutine ends inside the call;
+ // follow-up work belongs in terminal-coordinator callbacks, not after this invocation.
+ terminalCoordinator.complete(request = request, generateAi = ::generateAiSummaries)
}
/** Common cleanup path when a scan step fails mid-loop. */
- private suspend fun pauseAndAbort() {
- _scanState.value = DiscoveryScanState.Failed("Connection lost during scan")
- // pauseAndAbort runs inside the runScanLoop coroutine, which is a child of scanScope.
- // cancelScanInternal() cancels scanScope (and therefore this coroutine), so it must run LAST:
- // any suspend after it — finalizeSession or restoreHomePreset — would throw CancellationException
- // and silently skip cleanup, stranding the radio on the scan modem preset. So finalize and reach
- // the terminal state first, then restore the home preset on applicationScope (which outlives
- // scanScope), mirroring stopScan().
- finalizeSession("failed")
- _scanState.value = DiscoveryScanState.Complete(DiscoveryScanState.CompletionOutcome.Failed)
- applicationScope.launch { restoreHomePreset() }
- cancelScanInternal()
+ private suspend fun pauseAndAbort(persistPartialDwell: Boolean = false) {
+ val request =
+ mutex.withLock {
+ if (_scanState.value is DiscoveryScanState.Cancelling) {
+ null
+ } else {
+ _scanState.value = DiscoveryScanState.Analysis
+ terminalRequestLocked(
+ pendingStatus = DiscoverySessionStatus.RESTORE_PENDING_FAILED,
+ outcome = DiscoveryScanState.CompletionOutcome.Failed,
+ awaitRestore = false,
+ generateAi = false,
+ )
+ }
+ }
+ if (request == null) return
+ if (persistPartialDwell) {
+ terminalCoordinator.complete(request = request, beforeFinalize = ::persistCurrentDwellResults)
+ } else {
+ terminalCoordinator.complete(request = request)
+ }
}
private suspend fun shiftTarget(target: ScanTarget) {
@@ -394,8 +558,9 @@ class DiscoveryScanEngine(
),
),
)
+ currentCoroutineContext().ensureActive()
+ mutex.withLock { tunedPrimaryChannel = true }
radioController.setLocalChannel(Channel(index = 0, role = Channel.Role.PRIMARY, settings = target.channel))
- tunedPrimaryChannel = true
Logger.i { "DiscoveryScanEngine: shifted to ${target.label} (custom channel)" }
}
// The firmware often restarts the radio or reboots after a LoRa config change.
@@ -430,25 +595,37 @@ class DiscoveryScanEngine(
private suspend fun runDwell(presetName: String, durationSeconds: Long): Boolean {
var remaining = durationSeconds
- while (remaining > 0 && isActive) {
+ var canContinue = true
+ while (remaining > 0 && canContinue) {
val isConnected = serviceRepository.connectionState.value is ConnectionState.Connected
if (!isConnected) {
- _scanState.value = DiscoveryScanState.Reconnecting(presetName)
- val reconnected = waitForConnection()
- if (!reconnected) return false
- continue
+ canContinue = transitionScanState(DiscoveryScanState.Reconnecting(presetName)) && waitForConnection()
+ } else {
+ val dwellState =
+ DiscoveryScanState.Dwell(
+ presetName = presetName,
+ remainingSeconds = remaining,
+ totalSeconds = durationSeconds,
+ )
+ canContinue = transitionScanState(dwellState)
+ if (canContinue) {
+ delay(TICK_INTERVAL_MS)
+ remaining--
+ }
}
-
- _scanState.value =
- DiscoveryScanState.Dwell(
- presetName = presetName,
- remainingSeconds = remaining,
- totalSeconds = durationSeconds,
- )
- delay(TICK_INTERVAL_MS)
- remaining--
}
- return true
+ return canContinue
+ }
+
+ /** Caller must hold [mutex]. Terminal states are intentionally excluded even though Cancelling is active. */
+ private fun canAdvanceScanLocked(): Boolean = isActive &&
+ _scanState.value !is DiscoveryScanState.Cancelling &&
+ _scanState.value !is DiscoveryScanState.Analysis
+
+ private suspend fun transitionScanState(state: DiscoveryScanState): Boolean = mutex.withLock {
+ if (!canAdvanceScanLocked()) return@withLock false
+ _scanState.value = state
+ true
}
// endregion
@@ -532,13 +709,15 @@ class DiscoveryScanEngine(
private suspend fun persistCurrentDwellResults() {
if (sessionId == 0L) return
mutex.withLock {
+ if (currentDwellPersisted) return@withLock
if (collectedNodes.isEmpty()) {
persistEmptyPresetResult()
- return
+ currentDwellPersisted = true
+ } else {
+ val presetResultId = persistPresetResult()
+ persistDiscoveredNodes(presetResultId)
+ currentDwellPersisted = true
}
-
- val presetResultId = persistPresetResult()
- persistDiscoveredNodes(presetResultId)
}
}
@@ -673,116 +852,59 @@ class DiscoveryScanEngine(
return avgChannel to avgAirRate
}
- // Guarded by [mutex] so the session-row read-modify-write can't interleave with restoreInterruptedSessionIfAny().
- // pauseAndAbort() flips isActive to false before finalizing, so without this a reconnect-triggered restore could
- // race this write on the same row. No caller (runScanLoop, pauseAndAbort, stopScan) holds the mutex here.
- private suspend fun finalizeSession(status: String) = mutex.withLock {
- if (sessionId == 0L) return@withLock
- val uniqueCount = discoveryDao.getUniqueNodeCount(sessionId)
- val presetResults = discoveryDao.getPresetResults(sessionId)
- val session = discoveryDao.getSession(sessionId) ?: return@withLock
- val totalDwell = presetResults.sumOf { it.dwellDurationSeconds }
- val totalMsgs = presetResults.sumOf { it.messageCount }
- val totalSensor = presetResults.sumOf { it.sensorPacketCount }
- val maxDistance = discoveryDao.getMaxDistance(sessionId) ?: 0.0
- val avgChanUtil =
- presetResults
- .filter { it.uniqueNodes > 0 }
- .map { it.avgChannelUtilization }
- .average()
- .takeIf { !it.isNaN() } ?: 0.0
- discoveryDao.updateSession(
- session.copy(
- totalUniqueNodes = uniqueCount,
- totalDwellSeconds = totalDwell,
- totalMessages = totalMsgs,
- totalSensorPackets = totalSensor,
- furthestNodeDistance = maxDistance,
- avgChannelUtilization = avgChanUtil,
- completionStatus = status,
- ),
- )
- _currentSession.value = discoveryDao.getSession(sessionId)
- }
-
- // endregion
-
- // region Home preset restoration
-
- private suspend fun restoreHomePreset() {
- val config = originalLoRaConfig ?: return
- // Restore the primary channel first (only when a custom-channel target overwrote it), then the LoRa config —
- // both are no-ops on a public-only scan, so that path stays unchanged (FR-005 no-regression).
- if (tunedPrimaryChannel) {
- originalPrimaryChannel?.let {
- radioController.setLocalChannel(Channel(index = 0, role = Channel.Role.PRIMARY, settings = it))
+ /** Builds one immutable terminal snapshot. Caller must hold [mutex]. */
+ private fun terminalRequestLocked(
+ pendingStatus: String,
+ outcome: DiscoveryScanState.CompletionOutcome,
+ awaitRestore: Boolean,
+ generateAi: Boolean,
+ ): DiscoveryTerminalRequest {
+ val session = _currentSession.value
+ val finalStatus = finalStatusForPendingStatus(pendingStatus)
+ val config = session?.homeLoraConfig ?: originalLoRaConfig
+ val primaryChannel = session?.homePrimaryChannel ?: originalPrimaryChannel
+ if (tunedPrimaryChannel && primaryChannel == null) {
+ Logger.e {
+ "DiscoveryScanEngine: primary channel was retuned without a captured home channel; " +
+ "restoring LoRa config only for session ${session?.id ?: sessionId}"
}
}
- radioController.setLocalConfig(Config(lora = config))
- Logger.i { "DiscoveryScanEngine: restored original LoRa config" }
- // The firmware often restarts the radio or reboots after a LoRa config change.
- delay(3000)
- // Wait briefly for reconnection after restoring
- waitForConnection()
+ val restorePlan =
+ if (session == null || config == null) {
+ null
+ } else {
+ DiscoveryHomeRestorePlan(
+ sessionId = session.id,
+ deviceAddress = session.deviceAddress,
+ loraConfig = config,
+ primaryChannel = primaryChannel,
+ restorePrimaryChannel = tunedPrimaryChannel && primaryChannel != null,
+ finalStatus = finalStatus,
+ )
+ }
+ return DiscoveryTerminalRequest(
+ sessionId = session?.id ?: sessionId,
+ restorePlan = restorePlan,
+ pendingStatus = pendingStatus,
+ outcome = outcome,
+ awaitRestore = awaitRestore,
+ shouldGenerateAi = generateAi,
+ )
}
+ private fun finalStatusForPendingStatus(pendingStatus: String): String =
+ finalStatusForPendingRestore(pendingStatus, default = DiscoverySessionStatus.FAILED)
+
// endregion
// region Interrupted-session restoration
/**
- * Watches for the radio reconnecting and restores any home LoRa config a *previous process's* scan left mid-flight
- * — process death, a crash, or BLE loss mid-scan skips [restoreHomePreset] entirely, leaving the session row
- * "in_progress" forever and the radio detuned off the user's home mesh until this catches it.
- *
- * Meant to be `launch`ed once for the app's process lifetime (from app startup) — it never returns normally. Safe
- * to call even while this engine is running a legitimate scan of its own: the [isActive] check (taken under
- * [mutex], the same lock [startScanTargets] uses) skips every reconnect that belongs to this process's own
- * in-progress scan.
- *
- * @param onRestored invoked with the restored session's home-preset label after a successful restore. The engine
- * stays UI-free: localizing that into a user notification is the caller's job (see MeshUtilApplication), which
- * also keeps this function unit-testable off a live Compose-resources runtime.
+ * Watches reconnects for persisted discovery sessions left without a confirmed home-radio restore. Recovery is
+ * device-bound and skipped while this engine owns an active scan.
*/
- suspend fun restoreInterruptedSessionsOnReconnect(onRestored: suspend (homePreset: String) -> Unit = {}) {
- serviceRepository.connectionState.collect { state ->
- if (state !is ConnectionState.Connected) return@collect
- try {
- restoreInterruptedSessionIfAny()?.let { onRestored(it) }
- } catch (e: CancellationException) {
- throw e
- } catch (@Suppress("TooGenericExceptionCaught") e: Exception) {
- // A radio write can fail if the link drops right after Connected — the session stays
- // "interrupted" (it's only marked "restored" after the writes succeed), so the next
- // reconnect retries. Swallowing keeps this process-lifetime watcher alive.
- Logger.e(e) { "DiscoveryScanEngine: interrupted-session restore failed; will retry on reconnect" }
- }
- }
- }
-
- /** Restores an interrupted session's home config if one matches the current device; returns its home preset. */
- private suspend fun restoreInterruptedSessionIfAny(): String? {
- val address = meshPrefs.deviceAddress.value ?: return null
- return mutex.withLock {
- if (isActive) return@withLock null
- val session = discoveryDao.getInterruptedSession(address) ?: return@withLock null
- // A session with no captured config can never be restored (the radio config was null at scan start).
- // Terminalize it so it stops re-matching getInterruptedSession on every future reconnect.
- val loraConfig =
- session.homeLoraConfig
- ?: run {
- discoveryDao.updateSession(session.copy(completionStatus = "unrestorable"))
- return@withLock null
- }
- Logger.w { "DiscoveryScanEngine: restoring home config after interrupted session ${session.id}" }
- session.homePrimaryChannel?.let {
- radioController.setLocalChannel(Channel(index = 0, role = Channel.Role.PRIMARY, settings = it))
- }
- radioController.setLocalConfig(Config(lora = loraConfig))
- discoveryDao.updateSession(session.copy(completionStatus = "restored"))
- session.homePreset
- }
- }
+ suspend fun restoreInterruptedSessionsOnReconnect(onRestored: suspend (homePreset: String) -> Unit = {}): Unit =
+ interruptedSessionRecovery.watch(onRestored)
// endregion
@@ -799,6 +921,7 @@ class DiscoveryScanEngine(
// endregion
companion object {
+ private const val CONFIG_SNAPSHOT_TIMEOUT_MS = 5_000L
private const val RECONNECT_TIMEOUT_MS = 60_000L
private const val TICK_INTERVAL_MS = 1_000L
private const val POSITION_DIVISOR = 1e7
diff --git a/feature/discovery/src/commonMain/kotlin/org/meshtastic/feature/discovery/DiscoveryTerminalCoordinator.kt b/feature/discovery/src/commonMain/kotlin/org/meshtastic/feature/discovery/DiscoveryTerminalCoordinator.kt
new file mode 100644
index 0000000000..bfca9c9a80
--- /dev/null
+++ b/feature/discovery/src/commonMain/kotlin/org/meshtastic/feature/discovery/DiscoveryTerminalCoordinator.kt
@@ -0,0 +1,313 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.feature.discovery
+
+import co.touchlab.kermit.Logger
+import kotlinx.coroutines.CoroutineStart
+import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.NonCancellable
+import kotlinx.coroutines.async
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withContext
+import org.meshtastic.core.common.di.ApplicationCoroutineScope
+import org.meshtastic.core.common.util.safeCatching
+import org.meshtastic.core.database.dao.DiscoveryDao
+import org.meshtastic.core.database.entity.DiscoverySessionEntity
+import org.meshtastic.core.database.entity.DiscoverySessionStatus
+
+internal data class DiscoveryTerminalRequest(
+ val sessionId: Long,
+ val restorePlan: DiscoveryHomeRestorePlan?,
+ val pendingStatus: String,
+ val outcome: DiscoveryScanState.CompletionOutcome,
+ val awaitRestore: Boolean,
+ val shouldGenerateAi: Boolean,
+)
+
+/** Serializes terminal scan cleanup and keeps persistence separate from radio restoration ownership. */
+@Suppress("TooManyFunctions")
+internal class DiscoveryTerminalCoordinator(
+ private val discoveryDao: DiscoveryDao,
+ private val homeRestorer: DiscoveryHomeRestorer,
+ private val applicationScope: ApplicationCoroutineScope,
+ private val onSessionUpdated: (DiscoverySessionEntity) -> Unit,
+ private val onTerminalCompleted: (DiscoveryScanState.CompletionOutcome) -> Unit,
+ private val cancelScan: suspend () -> Unit,
+) {
+ private data class TerminalTaskSelection(
+ val task: Deferred<DiscoveryScanState.CompletionOutcome>,
+ val accepted: Boolean,
+ val ownerRequest: DiscoveryTerminalRequest,
+ )
+
+ private data class SessionAggregates(
+ val totalUniqueNodes: Int,
+ val totalDwellSeconds: Long,
+ val totalMessages: Int,
+ val totalSensorPackets: Int,
+ val furthestNodeDistance: Double,
+ val avgChannelUtilization: Double,
+ )
+
+ private val mutex = Mutex()
+ private var terminalCompletion: Deferred<DiscoveryScanState.CompletionOutcome>? = null
+ private var terminalRequest: DiscoveryTerminalRequest? = null
+ private var terminalRestoreTask: Deferred<Boolean>? = null
+ private var terminalRestoreSessionId: Long? = null
+
+ suspend fun resetForScan(): Boolean = mutex.withLock {
+ if (terminalCompletion?.isCompleted == false) {
+ Logger.w { "DiscoveryScanEngine: refusing reset while terminal cleanup is active" }
+ false
+ } else {
+ terminalCompletion = null
+ terminalRequest = null
+ terminalRestoreTask = null
+ terminalRestoreSessionId = null
+ true
+ }
+ }
+
+ suspend fun complete(
+ request: DiscoveryTerminalRequest,
+ beforeFinalize: suspend () -> Unit = {},
+ generateAi: suspend () -> Unit = {},
+ ): DiscoveryScanState.CompletionOutcome {
+ val selection = terminalTask(request, beforeFinalize, generateAi)
+ val sharedOutcome = selection.task.await()
+ // Joined callers with the same data-class value share the owner's work; changing request equality changes
+ // this deduplication contract.
+ if (selection.accepted || request == selection.ownerRequest) return sharedOutcome
+
+ Logger.w {
+ "DiscoveryScanEngine: terminal request ${request.pendingStatus} joined an active terminal cleanup; " +
+ "running its required follow-up work after the shared cleanup"
+ }
+ val joinedOutcome = runJoinedWork(request, beforeFinalize, generateAi, sharedOutcome)
+ publishJoinedOutcomeIfCurrent(selection.task, sharedOutcome, joinedOutcome)
+ return joinedOutcome
+ }
+
+ private suspend fun terminalTask(
+ request: DiscoveryTerminalRequest,
+ beforeFinalize: suspend () -> Unit,
+ generateAi: suspend () -> Unit,
+ ): TerminalTaskSelection {
+ val selection =
+ mutex.withLock {
+ val existing = terminalCompletion?.takeIf { !it.isCompleted }
+ if (existing != null) {
+ TerminalTaskSelection(existing, accepted = false, ownerRequest = checkNotNull(terminalRequest))
+ } else {
+ val newTask =
+ applicationScope.async(start = CoroutineStart.LAZY) {
+ var publishedOutcome = fallbackOutcome(request.outcome)
+ try {
+ runTerminalCleanup(request, beforeFinalize, generateAi).also { publishedOutcome = it }
+ } finally {
+ onTerminalCompleted(publishedOutcome)
+ }
+ }
+ terminalCompletion = newTask
+ terminalRequest = request
+ TerminalTaskSelection(newTask, accepted = true, ownerRequest = request)
+ }
+ }
+ // The task can acquire the scan-engine mutex through beforeFinalize; never start it under this mutex.
+ if (selection.accepted) selection.task.start()
+ return selection
+ }
+
+ private fun fallbackOutcome(requested: DiscoveryScanState.CompletionOutcome): DiscoveryScanState.CompletionOutcome =
+ if (requested == DiscoveryScanState.CompletionOutcome.Success) {
+ DiscoveryScanState.CompletionOutcome.Failed
+ } else {
+ requested
+ }
+
+ private suspend fun runTerminalCleanup(
+ request: DiscoveryTerminalRequest,
+ beforeFinalize: suspend () -> Unit,
+ generateAi: suspend () -> Unit,
+ ): DiscoveryScanState.CompletionOutcome {
+ val cancellationSucceeded = runBestEffort("scan cancellation failed during terminal cleanup", cancelScan)
+ val persistedStatus =
+ if (request.restorePlan == null) {
+ finalStatusForPendingRestore(request.pendingStatus, default = request.pendingStatus)
+ } else {
+ request.pendingStatus
+ }
+ var restoreTask: Deferred<Boolean>? = null
+ val persistenceSucceeded =
+ try {
+ val beforeFinalizeSucceeded =
+ runBestEffort("dwell persistence failed during terminal cleanup", beforeFinalize)
+ val terminalPersistSucceeded = persistTerminalSession(request.sessionId, persistedStatus)
+ cancellationSucceeded && beforeFinalizeSucceeded && terminalPersistSucceeded
+ } finally {
+ // Preserve aggregate-before-final-status ordering, but make restore scheduling cancellation-safe so a
+ // database cancellation cannot strand the radio on its scan configuration. Capture the task once so
+ // foreground waiting cannot accidentally schedule a second restore after a fast completion.
+ withContext(NonCancellable) {
+ restoreTask = request.restorePlan?.let { homeRestorer.schedule(it) }
+ mutex.withLock {
+ terminalRestoreTask = restoreTask
+ terminalRestoreSessionId = request.sessionId.takeIf { restoreTask != null }
+ }
+ }
+ }
+
+ var outcome = outcomeAfterPersistence(request.outcome, persistenceSucceeded)
+
+ if (request.awaitRestore && restoreTask != null && !awaitForegroundRestore(checkNotNull(restoreTask))) {
+ outcome = DiscoveryScanState.CompletionOutcome.Failed
+ homeRestorer.updateFinalStatus(request.sessionId, DiscoverySessionStatus.FAILED)
+ persistPendingRestoreStatus(request.sessionId, DiscoverySessionStatus.RESTORE_PENDING_FAILED)
+ }
+ if (request.shouldGenerateAi && outcome == DiscoveryScanState.CompletionOutcome.Success) {
+ runBestEffort("AI summary generation failed", generateAi)
+ }
+ return outcome
+ }
+
+ private suspend fun runJoinedWork(
+ request: DiscoveryTerminalRequest,
+ beforeFinalize: suspend () -> Unit,
+ generateAi: suspend () -> Unit,
+ sharedOutcome: DiscoveryScanState.CompletionOutcome,
+ ): DiscoveryScanState.CompletionOutcome {
+ val beforeFinalizeSucceeded = runBestEffort("joined dwell persistence failed", beforeFinalize)
+ val aggregateSucceeded = persistSessionAggregates(request.sessionId)
+ var outcome = outcomeAfterPersistence(sharedOutcome, beforeFinalizeSucceeded && aggregateSucceeded)
+ if (request.awaitRestore) {
+ val restoreTask =
+ mutex.withLock { terminalRestoreTask.takeIf { terminalRestoreSessionId == request.sessionId } }
+ if (restoreTask != null && !awaitForegroundRestore(restoreTask)) {
+ outcome = DiscoveryScanState.CompletionOutcome.Failed
+ homeRestorer.updateFinalStatus(request.sessionId, DiscoverySessionStatus.FAILED)
+ persistPendingRestoreStatus(request.sessionId, DiscoverySessionStatus.RESTORE_PENDING_FAILED)
+ }
+ }
+ if (request.shouldGenerateAi && outcome == DiscoveryScanState.CompletionOutcome.Success) {
+ runBestEffort("AI summary generation failed", generateAi)
+ }
+ return outcome
+ }
+
+ /** Republishes a corrected joined outcome only while the completed task still owns terminal state. */
+ private suspend fun publishJoinedOutcomeIfCurrent(
+ task: Deferred<DiscoveryScanState.CompletionOutcome>,
+ sharedOutcome: DiscoveryScanState.CompletionOutcome,
+ joinedOutcome: DiscoveryScanState.CompletionOutcome,
+ ) {
+ if (joinedOutcome == sharedOutcome) return
+ mutex.withLock { if (terminalCompletion === task) onTerminalCompleted(joinedOutcome) }
+ }
+
+ private fun outcomeAfterPersistence(
+ requested: DiscoveryScanState.CompletionOutcome,
+ persistenceSucceeded: Boolean,
+ ): DiscoveryScanState.CompletionOutcome =
+ if (!persistenceSucceeded && requested == DiscoveryScanState.CompletionOutcome.Success) {
+ DiscoveryScanState.CompletionOutcome.Failed
+ } else {
+ requested
+ }
+
+ private suspend fun runBestEffort(message: String, block: suspend () -> Unit): Boolean {
+ val result = safeCatching { block() }
+ val failure = result.exceptionOrNull()
+ if (failure != null) {
+ Logger.e(failure) { "DiscoveryScanEngine: $message" }
+ }
+ return failure == null
+ }
+
+ private suspend fun persistPendingRestoreStatus(sessionId: Long, status: String): Boolean {
+ if (sessionId == 0L) return true
+ return persistAndPublish("pending restore status persistence failed") {
+ discoveryDao.updateRecoverableSessionCompletionStatus(sessionId, status)
+ discoveryDao.getSession(sessionId)
+ }
+ }
+
+ private suspend fun persistTerminalSession(sessionId: Long, status: String): Boolean {
+ if (sessionId == 0L) return true
+ return persistAndPublish("terminal session persistence failed") {
+ val aggregates = calculateSessionAggregates(sessionId) ?: return@persistAndPublish null
+ updateSessionAggregates(sessionId, aggregates)
+ discoveryDao.updateRecoverableSessionCompletionStatus(sessionId, status)
+ discoveryDao.getSession(sessionId)
+ }
+ }
+
+ private suspend fun persistSessionAggregates(sessionId: Long): Boolean {
+ if (sessionId == 0L) return true
+ return persistAndPublish("terminal aggregate persistence failed") {
+ val aggregates = calculateSessionAggregates(sessionId) ?: return@persistAndPublish null
+ updateSessionAggregates(sessionId, aggregates)
+ discoveryDao.getSession(sessionId)
+ }
+ }
+
+ private suspend fun persistAndPublish(
+ failureMessage: String,
+ update: suspend () -> DiscoverySessionEntity?,
+ ): Boolean {
+ val result = safeCatching { update() }
+ val failure = result.exceptionOrNull()
+ if (failure != null) Logger.e(failure) { "DiscoveryScanEngine: $failureMessage" }
+ val updated = result.getOrNull()
+ if (failure == null && updated == null) {
+ Logger.e { "DiscoveryScanEngine: $failureMessage; the session row was not found" }
+ }
+ updated?.let(onSessionUpdated)
+ return updated != null
+ }
+
+ private suspend fun calculateSessionAggregates(sessionId: Long): SessionAggregates? {
+ if (discoveryDao.getSession(sessionId) == null) return null
+ val presetResults = discoveryDao.getPresetResults(sessionId)
+ val avgChannelUtilization =
+ presetResults
+ .filter { it.uniqueNodes > 0 }
+ .map { it.avgChannelUtilization }
+ .average()
+ .takeIf { !it.isNaN() } ?: 0.0
+ return SessionAggregates(
+ totalUniqueNodes = discoveryDao.getUniqueNodeCount(sessionId),
+ totalDwellSeconds = presetResults.sumOf { it.dwellDurationSeconds },
+ totalMessages = presetResults.sumOf { it.messageCount },
+ totalSensorPackets = presetResults.sumOf { it.sensorPacketCount },
+ furthestNodeDistance = discoveryDao.getMaxDistance(sessionId) ?: 0.0,
+ avgChannelUtilization = avgChannelUtilization,
+ )
+ }
+
+ private suspend fun updateSessionAggregates(sessionId: Long, aggregates: SessionAggregates) {
+ discoveryDao.updateSessionAggregates(
+ sessionId = sessionId,
+ totalUniqueNodes = aggregates.totalUniqueNodes,
+ totalDwellSeconds = aggregates.totalDwellSeconds,
+ totalMessages = aggregates.totalMessages,
+ totalSensorPackets = aggregates.totalSensorPackets,
+ furthestNodeDistance = aggregates.furthestNodeDistance,
+ avgChannelUtilization = aggregates.avgChannelUtilization,
+ )
+ }
+}
diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryHistoryBehaviorTest.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryHistoryBehaviorTest.kt
index 1947ab4d61..865d4027f5 100644
--- a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryHistoryBehaviorTest.kt
+++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryHistoryBehaviorTest.kt
@@ -18,16 +18,12 @@
package org.meshtastic.feature.discovery
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.first
-import kotlinx.coroutines.flow.flowOf
-import kotlinx.coroutines.flow.update
import kotlinx.coroutines.test.runTest
-import org.meshtastic.core.database.dao.DiscoveryDao
import org.meshtastic.core.database.entity.DiscoveredNodeEntity
import org.meshtastic.core.database.entity.DiscoveryPresetResultEntity
import org.meshtastic.core.database.entity.DiscoverySessionEntity
+import org.meshtastic.core.database.entity.DiscoverySessionStatus
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertNotNull
@@ -37,7 +33,7 @@ import kotlin.test.assertTrue
/** Tests for session history: sorting, session load by ID, and delete behavior (D042). */
class DiscoveryHistoryBehaviorTest {
- private val dao = HistoryTestDao()
+ private val dao = SharedInMemoryDiscoveryDao()
// region History sorting
@@ -149,121 +145,8 @@ class DiscoveryHistoryBehaviorTest {
timestamp = timestamp,
presetsScanned = "LONG_FAST",
homePreset = homePreset,
- completionStatus = "complete",
+ completionStatus = DiscoverySessionStatus.COMPLETE,
)
// endregion
}
-
-// region In-memory DAO for history tests
-
-private class HistoryTestDao : DiscoveryDao {
- private var nextSessionId = 1L
- private var nextPresetResultId = 1L
- private var nextNodeId = 1L
-
- private val sessions = mutableMapOf<Long, DiscoverySessionEntity>()
- private val presetResults = mutableMapOf<Long, DiscoveryPresetResultEntity>()
- private val discoveredNodes = mutableMapOf<Long, DiscoveredNodeEntity>()
- private val sessionsFlow = MutableStateFlow<List<DiscoverySessionEntity>>(emptyList())
-
- private fun refreshSessionsFlow() {
- sessionsFlow.update { sessions.values.sortedByDescending { it.timestamp } }
- }
-
- override suspend fun insertSession(session: DiscoverySessionEntity): Long {
- val id = nextSessionId++
- sessions[id] = session.copy(id = id)
- refreshSessionsFlow()
- return id
- }
-
- override suspend fun updateSession(session: DiscoverySessionEntity) {
- sessions[session.id] = session
- refreshSessionsFlow()
- }
-
- override fun getAllSessions(): Flow<List<DiscoverySessionEntity>> = sessionsFlow
-
- override suspend fun getAllSessionsSnapshot(): List<DiscoverySessionEntity> = sessions.values.toList()
-
- override suspend fun getSession(sessionId: Long) = sessions[sessionId]
-
- override fun getSessionFlow(sessionId: Long): Flow<DiscoverySessionEntity?> = MutableStateFlow(sessions[sessionId])
-
- override suspend fun deleteSession(sessionId: Long) {
- sessions.remove(sessionId)
- val resultIds = presetResults.values.filter { it.sessionId == sessionId }.map { it.id }
- resultIds.forEach { rid ->
- discoveredNodes.entries.removeAll { it.value.presetResultId == rid }
- presetResults.remove(rid)
- }
- refreshSessionsFlow()
- }
-
- override suspend fun insertPresetResult(result: DiscoveryPresetResultEntity): Long {
- val id = nextPresetResultId++
- presetResults[id] = result.copy(id = id)
- return id
- }
-
- override suspend fun updatePresetResult(result: DiscoveryPresetResultEntity) {
- presetResults[result.id] = result
- }
-
- override suspend fun getPresetResults(sessionId: Long) = presetResults.values.filter { it.sessionId == sessionId }
-
- override fun getPresetResultsFlow(sessionId: Long) =
- flowOf(presetResults.values.filter { it.sessionId == sessionId })
-
- override suspend fun insertDiscoveredNode(node: DiscoveredNodeEntity): Long {
- val id = nextNodeId++
- discoveredNodes[id] = node.copy(id = id)
- return id
- }
-
- override suspend fun insertDiscoveredNodes(nodes: List<DiscoveredNodeEntity>) {
- nodes.forEach { insertDiscoveredNode(it) }
- }
-
- override suspend fun updateDiscoveredNode(node: DiscoveredNodeEntity) {
- discoveredNodes[node.id] = node
- }
-
- override suspend fun getDiscoveredNodes(presetResultId: Long) =
- discoveredNodes.values.filter { it.presetResultId == presetResultId }
-
- override fun getDiscoveredNodesFlow(presetResultId: Long) =
- flowOf(discoveredNodes.values.filter { it.presetResultId == presetResultId })
-
- override suspend fun getUniqueNodeNums(sessionId: Long) = presetResults.values
- .filter { it.sessionId == sessionId }
- .flatMap { pr -> discoveredNodes.values.filter { it.presetResultId == pr.id } }
- .map { it.nodeNum }
- .distinct()
-
- override suspend fun getUniqueNodeCount(sessionId: Long) = getUniqueNodeNums(sessionId).size
-
- override suspend fun getMaxDistance(sessionId: Long) = presetResults.values
- .filter { it.sessionId == sessionId }
- .flatMap { pr -> discoveredNodes.values.filter { it.presetResultId == pr.id } }
- .mapNotNull { it.distanceFromUser }
- .maxOrNull()
-
- override suspend fun getSessionWithResults(sessionId: Long) = sessions[sessionId]
-
- override suspend fun markInterruptedSessions() {
- sessions.keys.toList().forEach { key ->
- val session = sessions[key]!!
- if (session.completionStatus == "in_progress") {
- sessions[key] = session.copy(completionStatus = "interrupted")
- }
- }
- }
-
- override suspend fun getInterruptedSession(deviceAddress: String): DiscoverySessionEntity? = sessions.values
- .filter { it.deviceAddress == deviceAddress && it.completionStatus in setOf("in_progress", "interrupted") }
- .maxByOrNull { it.timestamp }
-}
-
-// endregion
diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryHomeRestorerTest.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryHomeRestorerTest.kt
new file mode 100644
index 0000000000..8da4487119
--- /dev/null
+++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryHomeRestorerTest.kt
@@ -0,0 +1,774 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.feature.discovery
+
+import dev.mokkery.MockMode
+import dev.mokkery.mock
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.async
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.advanceTimeBy
+import kotlinx.coroutines.test.advanceUntilIdle
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import org.meshtastic.core.common.di.ApplicationCoroutineScope
+import org.meshtastic.core.database.dao.DiscoveryDao
+import org.meshtastic.core.database.entity.DiscoverySessionEntity
+import org.meshtastic.core.database.entity.DiscoverySessionStatus
+import org.meshtastic.core.model.ConnectionState
+import org.meshtastic.core.testing.FakeMeshPrefs
+import org.meshtastic.core.testing.FakeServiceRepository
+import org.meshtastic.proto.Channel
+import org.meshtastic.proto.Config
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertNotSame
+import kotlin.test.assertSame
+import kotlin.test.assertTrue
+
+class DiscoveryHomeRestorerTest {
+ private fun applicationScope(scope: CoroutineScope): ApplicationCoroutineScope =
+ object : ApplicationCoroutineScope {
+ override val coroutineContext = scope.coroutineContext
+ }
+
+ @Test
+ fun supersededRestoreReturnsFalseToForegroundWaiterWithoutCancellingIt() = runTest {
+ val firstDevice = "x:FIRST"
+ val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress(firstDevice) }
+ val serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Disconnected) }
+ val appScope = applicationScope(backgroundScope)
+ val restorer =
+ DiscoveryHomeRestorer(
+ radioController = DiscoveryTestRadioController(),
+ serviceRepository = serviceRepository,
+ discoveryDao = mock<DiscoveryDao>(MockMode.autofill),
+ applicationScope = appScope,
+ meshPrefs = meshPrefs,
+ )
+ val plan =
+ DiscoveryHomeRestorePlan(
+ sessionId = 1L,
+ deviceAddress = firstDevice,
+ loraConfig = Config.LoRaConfig(use_preset = true),
+ primaryChannel = null,
+ restorePrimaryChannel = false,
+ finalStatus = DiscoverySessionStatus.COMPLETE,
+ )
+
+ val foregroundWaiter = async { restorer.awaitForeground(plan) }
+ runCurrent()
+ assertFalse(foregroundWaiter.isCompleted, "the restore should be waiting for the disconnected device")
+
+ assertTrue(restorer.awaitBeforeScan("x:SECOND"), "a different device may proceed after superseding the restore")
+ assertFalse(
+ foregroundWaiter.await(),
+ "superseding a restore must return false instead of cancelling its waiter",
+ )
+ }
+
+ @Test
+ fun selectedDeviceChangeWakesDisconnectedRestoreWaiter() = runTest {
+ val device = "x:FIRST"
+ val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress(device) }
+ val serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Disconnected) }
+ val restorer =
+ DiscoveryHomeRestorer(
+ radioController = DiscoveryTestRadioController(),
+ serviceRepository = serviceRepository,
+ discoveryDao = mock<DiscoveryDao>(MockMode.autofill),
+ applicationScope = applicationScope(backgroundScope),
+ meshPrefs = meshPrefs,
+ )
+ val plan =
+ DiscoveryHomeRestorePlan(
+ sessionId = 1L,
+ deviceAddress = device,
+ loraConfig = Config.LoRaConfig(use_preset = true),
+ primaryChannel = null,
+ restorePrimaryChannel = false,
+ finalStatus = DiscoverySessionStatus.COMPLETE,
+ )
+
+ val result = restorer.schedule(plan)
+ runCurrent()
+ assertFalse(result.isCompleted)
+
+ meshPrefs.setDeviceAddress("x:SECOND")
+ runCurrent()
+
+ assertFalse(result.await())
+ }
+
+ @Test
+ fun finalStatusMappingPreservesTerminalIntentAndDefaultsRecoveredStates() {
+ assertEquals(
+ DiscoverySessionStatus.COMPLETE,
+ finalStatusForPendingRestore(DiscoverySessionStatus.RESTORE_PENDING_COMPLETE),
+ )
+ assertEquals(
+ DiscoverySessionStatus.STOPPED,
+ finalStatusForPendingRestore(DiscoverySessionStatus.RESTORE_PENDING_STOPPED),
+ )
+ assertEquals(
+ DiscoverySessionStatus.FAILED,
+ finalStatusForPendingRestore(DiscoverySessionStatus.RESTORE_PENDING_FAILED),
+ )
+ assertEquals(DiscoverySessionStatus.RESTORED, finalStatusForPendingRestore(DiscoverySessionStatus.IN_PROGRESS))
+ assertEquals(DiscoverySessionStatus.RESTORED, finalStatusForPendingRestore(DiscoverySessionStatus.INTERRUPTED))
+ assertEquals(
+ DiscoverySessionStatus.FAILED,
+ finalStatusForPendingRestore(DiscoverySessionStatus.INTERRUPTED, default = DiscoverySessionStatus.FAILED),
+ )
+ }
+
+ @Test
+ fun repeatedScheduleForSameActiveSessionIsIdempotent() = runTest {
+ val device = "x:SAME"
+ val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress(device) }
+ val serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Disconnected) }
+ val appScope = applicationScope(backgroundScope)
+ val restorer =
+ DiscoveryHomeRestorer(
+ radioController = DiscoveryTestRadioController(),
+ serviceRepository = serviceRepository,
+ discoveryDao = mock<DiscoveryDao>(MockMode.autofill),
+ applicationScope = appScope,
+ meshPrefs = meshPrefs,
+ )
+ val plan =
+ DiscoveryHomeRestorePlan(
+ sessionId = 7L,
+ deviceAddress = device,
+ loraConfig = Config.LoRaConfig(use_preset = true),
+ primaryChannel = null,
+ restorePrimaryChannel = false,
+ finalStatus = DiscoverySessionStatus.COMPLETE,
+ )
+
+ val first = restorer.schedule(plan)
+ val second = restorer.schedule(plan)
+
+ assertSame(first, second)
+ first.cancel()
+ }
+
+ @Test
+ fun schedulingDifferentSessionCancelsAndReplacesPendingRestore() = runTest {
+ val device = "x:SUPERSEDE"
+ val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress(device) }
+ val serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Disconnected) }
+ val restorer =
+ DiscoveryHomeRestorer(
+ radioController = DiscoveryTestRadioController(),
+ serviceRepository = serviceRepository,
+ discoveryDao = mock<DiscoveryDao>(MockMode.autofill),
+ applicationScope = applicationScope(backgroundScope),
+ meshPrefs = meshPrefs,
+ )
+ val firstPlan =
+ DiscoveryHomeRestorePlan(
+ sessionId = 1L,
+ deviceAddress = device,
+ loraConfig = Config.LoRaConfig(use_preset = true),
+ primaryChannel = null,
+ restorePrimaryChannel = false,
+ finalStatus = DiscoverySessionStatus.COMPLETE,
+ )
+
+ val first = restorer.schedule(firstPlan)
+ val second = restorer.schedule(firstPlan.copy(sessionId = 2L))
+
+ assertTrue(first.isCancelled, "the superseded session must release its restore ownership")
+ assertNotSame(first, second)
+ second.cancel()
+ }
+
+ @Test
+ fun sameSessionIdOnDifferentDeviceReplacesPendingRestore() = runTest {
+ val firstDevice = "x:FIRST"
+ val secondDevice = "x:SECOND"
+ val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress(firstDevice) }
+ val restorer =
+ DiscoveryHomeRestorer(
+ radioController = DiscoveryTestRadioController(),
+ serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Disconnected) },
+ discoveryDao = mock<DiscoveryDao>(MockMode.autofill),
+ applicationScope = applicationScope(backgroundScope),
+ meshPrefs = meshPrefs,
+ )
+ val firstPlan =
+ DiscoveryHomeRestorePlan(
+ sessionId = 7L,
+ deviceAddress = firstDevice,
+ loraConfig = Config.LoRaConfig(use_preset = true),
+ primaryChannel = null,
+ restorePrimaryChannel = false,
+ finalStatus = DiscoverySessionStatus.COMPLETE,
+ )
+
+ val first = restorer.schedule(firstPlan)
+ meshPrefs.setDeviceAddress(secondDevice)
+ val second = restorer.schedule(firstPlan.copy(deviceAddress = secondDevice))
+
+ assertTrue(first.isCancelled, "a reused database row id must not retain the previous radio's ownership")
+ assertNotSame(first, second)
+ second.cancel()
+ }
+
+ @Test
+ fun interruptedRecoverySwitchesActiveWatcherToNewlySelectedDevice() = runTest {
+ val firstDevice = "x:FIRST"
+ val secondDevice = "x:SECOND"
+ val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress(firstDevice) }
+ val serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Connected) }
+ val discoveryDao = SharedInMemoryDiscoveryDao()
+ listOf(firstDevice to "FIRST_HOME", secondDevice to "SECOND_HOME").forEach { (device, homePreset) ->
+ discoveryDao.insertSession(
+ DiscoverySessionEntity(
+ timestamp = 1L,
+ presetsScanned = "SHORT_FAST",
+ homePreset = homePreset,
+ completionStatus = DiscoverySessionStatus.INTERRUPTED,
+ deviceAddress = device,
+ homeLoraConfig = Config.LoRaConfig(use_preset = true),
+ ),
+ )
+ }
+ val firstRestore = CompletableDeferred<Boolean>()
+ val scheduledDevices = mutableListOf<String>()
+ val recovery =
+ DiscoveryInterruptedSessionRecovery(
+ serviceRepository = serviceRepository,
+ discoveryDao = discoveryDao,
+ meshPrefs = meshPrefs,
+ isScanActive = { false },
+ scheduleRestoreIfIdle = { session ->
+ scheduledDevices += checkNotNull(session.deviceAddress)
+ if (session.deviceAddress == firstDevice) firstRestore else CompletableDeferred(true)
+ },
+ )
+ val restoredPresets = mutableListOf<String>()
+ val watcher = launch { recovery.watch { restoredPresets += it } }
+ runCurrent()
+
+ meshPrefs.setDeviceAddress(secondDevice)
+ runCurrent()
+
+ assertEquals(listOf(firstDevice, secondDevice), scheduledDevices)
+ assertEquals(listOf("SECOND_HOME"), restoredPresets)
+ watcher.cancel()
+ firstRestore.cancel()
+ }
+
+ @Test
+ fun discoveryTestRadioControllerRejectsStaleDeviceBeforeWriting() = runTest {
+ val radioController = DiscoveryTestRadioController().apply { selectedDeviceAddress = "x:CURRENT" }
+
+ val restored =
+ radioController.restoreLocalConfiguration(
+ expectedDeviceAddress = "x:STALE",
+ config = Config(lora = Config.LoRaConfig(use_preset = true)),
+ primaryChannel = Channel(index = 0),
+ )
+
+ assertFalse(restored)
+ assertTrue(radioController.channelWrites.isEmpty())
+ assertTrue(radioController.configWrites.isEmpty())
+ }
+
+ @Test
+ fun unsatisfiablePrimaryChannelRestoreStopsWithoutRetrying() = runTest {
+ val device = "x:INVALID"
+ val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress(device) }
+ val radioController = DiscoveryTestRadioController()
+ val discoveryDao = SharedInMemoryDiscoveryDao()
+ val sessionId =
+ discoveryDao.insertSession(
+ DiscoverySessionEntity(
+ timestamp = 1L,
+ presetsScanned = "LONG_FAST",
+ homePreset = "LONG_FAST",
+ completionStatus = DiscoverySessionStatus.RESTORE_PENDING_FAILED,
+ deviceAddress = device,
+ ),
+ )
+ val restorer =
+ DiscoveryHomeRestorer(
+ radioController = radioController,
+ serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Connected) },
+ discoveryDao = discoveryDao,
+ applicationScope = applicationScope(backgroundScope),
+ meshPrefs = meshPrefs,
+ )
+ val plan =
+ DiscoveryHomeRestorePlan(
+ sessionId = sessionId,
+ deviceAddress = device,
+ loraConfig = Config.LoRaConfig(use_preset = true),
+ primaryChannel = null,
+ restorePrimaryChannel = true,
+ finalStatus = DiscoverySessionStatus.FAILED,
+ )
+
+ val result = restorer.schedule(plan)
+ runCurrent()
+
+ assertTrue(result.isCompleted, "an invalid restore plan must terminate without entering the retry loop")
+ assertFalse(result.await())
+ assertTrue(radioController.configWrites.isEmpty())
+ assertEquals(DiscoverySessionStatus.UNRESTORABLE, discoveryDao.getSession(sessionId)?.completionStatus)
+ assertFalse(restorer.hasUnsatisfiedRestoreFor(device))
+ }
+
+ @Test
+ fun transientWriteFailuresRetryWithBackoffWhileTheSameDeviceRemainsConnected() = runTest {
+ val device = "x:RETRY"
+ val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress(device) }
+ val serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Connected) }
+ val radioController =
+ DiscoveryTestRadioController().apply {
+ selectedDeviceAddress = device
+ failLocalConfigWritesRemaining = 2
+ }
+ val discoveryDao = SharedInMemoryDiscoveryDao()
+ val loraConfig = Config.LoRaConfig(use_preset = true)
+ val sessionId =
+ discoveryDao.insertSession(
+ DiscoverySessionEntity(
+ timestamp = 1L,
+ presetsScanned = "LONG_FAST",
+ homePreset = "LONG_FAST",
+ completionStatus = DiscoverySessionStatus.RESTORE_PENDING_COMPLETE,
+ deviceAddress = device,
+ homeLoraConfig = loraConfig,
+ ),
+ )
+ val restorer =
+ DiscoveryHomeRestorer(
+ radioController = radioController,
+ serviceRepository = serviceRepository,
+ discoveryDao = discoveryDao,
+ applicationScope = applicationScope(backgroundScope),
+ meshPrefs = meshPrefs,
+ )
+ val plan =
+ DiscoveryHomeRestorePlan(
+ sessionId = sessionId,
+ deviceAddress = device,
+ loraConfig = loraConfig,
+ primaryChannel = null,
+ restorePrimaryChannel = false,
+ finalStatus = DiscoverySessionStatus.COMPLETE,
+ )
+
+ val result = restorer.schedule(plan)
+ runCurrent()
+ assertFalse(result.isCompleted)
+ assertEquals(0, radioController.configWrites.size)
+
+ advanceTimeBy(DiscoveryHomeRestorer.RETRY_DELAY_MS)
+ runCurrent()
+ assertEquals(0, radioController.configWrites.size)
+ advanceTimeBy(DiscoveryHomeRestorer.RETRY_DELAY_MS)
+ runCurrent()
+ assertEquals(0, radioController.configWrites.size, "the second retry must use exponential backoff")
+ advanceTimeBy(DiscoveryHomeRestorer.RETRY_DELAY_MS)
+ runCurrent()
+ assertEquals(1, radioController.configWrites.size)
+ advanceTimeBy(DiscoveryHomeRestorer.POST_RESTORE_SETTLE_DELAY_MS)
+ runCurrent()
+
+ assertTrue(result.await())
+ assertEquals(DiscoverySessionStatus.COMPLETE, discoveryDao.getSession(sessionId)?.completionStatus)
+ }
+
+ @Test
+ fun updateFinalStatusChangesAndCorrectsTheTerminalStatusPublishedByARestore() = runTest {
+ val device = "x:STATUS"
+ val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress(device) }
+ val serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Connected) }
+ val radioController = DiscoveryTestRadioController().apply { selectedDeviceAddress = device }
+ val discoveryDao = SharedInMemoryDiscoveryDao()
+ val loraConfig = Config.LoRaConfig(use_preset = true)
+ val sessionId =
+ discoveryDao.insertSession(
+ DiscoverySessionEntity(
+ timestamp = 1L,
+ presetsScanned = "LONG_FAST",
+ homePreset = "LONG_FAST",
+ completionStatus = DiscoverySessionStatus.RESTORE_PENDING_COMPLETE,
+ deviceAddress = device,
+ homeLoraConfig = loraConfig,
+ ),
+ )
+ val restorer =
+ DiscoveryHomeRestorer(
+ radioController = radioController,
+ serviceRepository = serviceRepository,
+ discoveryDao = discoveryDao,
+ applicationScope = applicationScope(backgroundScope),
+ meshPrefs = meshPrefs,
+ )
+ val plan =
+ DiscoveryHomeRestorePlan(
+ sessionId = sessionId,
+ deviceAddress = device,
+ loraConfig = loraConfig,
+ primaryChannel = null,
+ restorePrimaryChannel = false,
+ finalStatus = DiscoverySessionStatus.COMPLETE,
+ )
+
+ val result = restorer.schedule(plan)
+ runCurrent()
+ assertFalse(result.isCompleted)
+ restorer.updateFinalStatus(sessionId, DiscoverySessionStatus.FAILED)
+ advanceTimeBy(DiscoveryHomeRestorer.POST_RESTORE_SETTLE_DELAY_MS)
+ runCurrent()
+
+ assertTrue(result.await())
+ assertEquals(DiscoverySessionStatus.FAILED, discoveryDao.getSession(sessionId)?.completionStatus)
+
+ restorer.updateFinalStatus(sessionId, DiscoverySessionStatus.STOPPED)
+
+ assertEquals(DiscoverySessionStatus.STOPPED, discoveryDao.getSession(sessionId)?.completionStatus)
+ }
+
+ @Test
+ @Suppress("LongMethod")
+ fun ownershipRejectionDoesNotConsumeTheRestoreRetryBudget() = runTest {
+ val device = "x:OWNERSHIP-LAG"
+ val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress(device) }
+ val serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Connected) }
+ val radioController =
+ DiscoveryTestRadioController().apply {
+ selectedDeviceAddress = device
+ acceptConditionalRestore = false
+ }
+ val discoveryDao = SharedInMemoryDiscoveryDao()
+ val loraConfig = Config.LoRaConfig(use_preset = true)
+ val sessionId =
+ discoveryDao.insertSession(
+ DiscoverySessionEntity(
+ timestamp = 1L,
+ presetsScanned = "LONG_FAST",
+ homePreset = "LONG_FAST",
+ completionStatus = DiscoverySessionStatus.RESTORE_PENDING_COMPLETE,
+ deviceAddress = device,
+ homeLoraConfig = loraConfig,
+ ),
+ )
+ val restorer =
+ DiscoveryHomeRestorer(
+ radioController = radioController,
+ serviceRepository = serviceRepository,
+ discoveryDao = discoveryDao,
+ applicationScope = applicationScope(backgroundScope),
+ meshPrefs = meshPrefs,
+ )
+ val result =
+ restorer.schedule(
+ DiscoveryHomeRestorePlan(
+ sessionId = sessionId,
+ deviceAddress = device,
+ loraConfig = loraConfig,
+ primaryChannel = null,
+ restorePrimaryChannel = false,
+ finalStatus = DiscoverySessionStatus.COMPLETE,
+ ),
+ )
+
+ advanceTimeBy(DiscoveryHomeRestorer.RETRY_DELAY_MS * (DiscoveryHomeRestorer.MAX_RESTORE_ATTEMPTS + 1))
+ runCurrent()
+
+ assertFalse(result.isCompleted, "ownership rejection must remain recoverable")
+ assertEquals(0, radioController.localConfigWriteAttempts)
+ assertEquals(
+ DiscoverySessionStatus.RESTORE_PENDING_COMPLETE,
+ discoveryDao.getSession(sessionId)?.completionStatus,
+ )
+
+ radioController.acceptConditionalRestore = true
+ advanceTimeBy(DiscoveryHomeRestorer.RETRY_DELAY_MS)
+ runCurrent()
+ advanceTimeBy(DiscoveryHomeRestorer.POST_RESTORE_SETTLE_DELAY_MS)
+ runCurrent()
+
+ assertTrue(result.await())
+ assertEquals(DiscoverySessionStatus.COMPLETE, discoveryDao.getSession(sessionId)?.completionStatus)
+ }
+
+ @Test
+ @Suppress("LongMethod")
+ fun persistentOwnershipRejectionStopsWithoutMarkingTheSessionUnrestorable() = runTest {
+ val device = "x:OWNERSHIP-STUCK"
+ val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress(device) }
+ val serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Connected) }
+ val radioController =
+ DiscoveryTestRadioController().apply {
+ selectedDeviceAddress = device
+ acceptConditionalRestore = false
+ }
+ val discoveryDao = SharedInMemoryDiscoveryDao()
+ val loraConfig = Config.LoRaConfig(use_preset = true)
+ val sessionId =
+ discoveryDao.insertSession(
+ DiscoverySessionEntity(
+ timestamp = 1L,
+ presetsScanned = "LONG_FAST",
+ homePreset = "LONG_FAST",
+ completionStatus = DiscoverySessionStatus.RESTORE_PENDING_COMPLETE,
+ deviceAddress = device,
+ homeLoraConfig = loraConfig,
+ ),
+ )
+ val restorer =
+ DiscoveryHomeRestorer(
+ radioController = radioController,
+ serviceRepository = serviceRepository,
+ discoveryDao = discoveryDao,
+ applicationScope = applicationScope(this),
+ meshPrefs = meshPrefs,
+ )
+
+ val result =
+ restorer.schedule(
+ DiscoveryHomeRestorePlan(
+ sessionId = sessionId,
+ deviceAddress = device,
+ loraConfig = loraConfig,
+ primaryChannel = null,
+ restorePrimaryChannel = false,
+ finalStatus = DiscoverySessionStatus.COMPLETE,
+ ),
+ )
+
+ advanceUntilIdle()
+
+ assertTrue(result.isCompleted, "persistent ownership rejection must not leave an unbounded retry task")
+ assertFalse(result.await())
+ assertEquals(0, radioController.localConfigWriteAttempts)
+ assertEquals(
+ DiscoverySessionStatus.RESTORE_PENDING_COMPLETE,
+ discoveryDao.getSession(sessionId)?.completionStatus,
+ "ownership rejection must stay recoverable instead of consuming the radio-write failure budget",
+ )
+ }
+
+ @Test
+ @Suppress("LongMethod")
+ fun transientDisconnectFailuresDoNotConsumeTheActiveRetryBudget() = runTest {
+ val device = "x:DISCONNECT-RETRY"
+ val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress(device) }
+ val serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Connected) }
+ val radioController =
+ DiscoveryTestRadioController().apply {
+ selectedDeviceAddress = device
+ failLocalConfigWritesRemaining = Int.MAX_VALUE
+ onLocalConfigWriteAttempt = { serviceRepository.setConnectionState(ConnectionState.Disconnected) }
+ }
+ val discoveryDao = SharedInMemoryDiscoveryDao()
+ val loraConfig = Config.LoRaConfig(use_preset = true)
+ val sessionId =
+ discoveryDao.insertSession(
+ DiscoverySessionEntity(
+ timestamp = 1L,
+ presetsScanned = "LONG_FAST",
+ homePreset = "LONG_FAST",
+ completionStatus = DiscoverySessionStatus.RESTORE_PENDING_COMPLETE,
+ deviceAddress = device,
+ homeLoraConfig = loraConfig,
+ ),
+ )
+ val restorer =
+ DiscoveryHomeRestorer(
+ radioController = radioController,
+ serviceRepository = serviceRepository,
+ discoveryDao = discoveryDao,
+ applicationScope = applicationScope(backgroundScope),
+ meshPrefs = meshPrefs,
+ )
+ val result =
+ restorer.schedule(
+ DiscoveryHomeRestorePlan(
+ sessionId = sessionId,
+ deviceAddress = device,
+ loraConfig = loraConfig,
+ primaryChannel = null,
+ restorePrimaryChannel = false,
+ finalStatus = DiscoverySessionStatus.COMPLETE,
+ ),
+ )
+
+ runCurrent()
+ repeat(DiscoveryHomeRestorer.MAX_RESTORE_ATTEMPTS + 1) {
+ assertFalse(result.isCompleted, "disconnect failures must remain recoverable")
+ assertEquals(
+ DiscoverySessionStatus.RESTORE_PENDING_COMPLETE,
+ discoveryDao.getSession(sessionId)?.completionStatus,
+ )
+ serviceRepository.setConnectionState(ConnectionState.Connected)
+ runCurrent()
+ }
+
+ radioController.failLocalConfigWritesRemaining = 0
+ radioController.onLocalConfigWriteAttempt = null
+ serviceRepository.setConnectionState(ConnectionState.Connected)
+ runCurrent()
+ advanceTimeBy(DiscoveryHomeRestorer.POST_RESTORE_SETTLE_DELAY_MS)
+ runCurrent()
+
+ assertTrue(result.await())
+ assertEquals(DiscoverySessionStatus.COMPLETE, discoveryDao.getSession(sessionId)?.completionStatus)
+ }
+
+ @Test
+ fun permanentWriteFailureStopsAfterBoundedAttemptsAndMarksSessionUnrestorable() = runTest {
+ val device = "x:UNRESTORABLE"
+ val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress(device) }
+ val serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Connected) }
+ val radioController =
+ DiscoveryTestRadioController().apply {
+ selectedDeviceAddress = device
+ failLocalConfigWritesRemaining = Int.MAX_VALUE
+ }
+ val discoveryDao = SharedInMemoryDiscoveryDao()
+ val loraConfig = Config.LoRaConfig(use_preset = true)
+ val sessionId =
+ discoveryDao.insertSession(
+ DiscoverySessionEntity(
+ timestamp = 1L,
+ presetsScanned = "LONG_FAST",
+ homePreset = "LONG_FAST",
+ completionStatus = DiscoverySessionStatus.RESTORE_PENDING_FAILED,
+ deviceAddress = device,
+ homeLoraConfig = loraConfig,
+ ),
+ )
+ val restorer =
+ DiscoveryHomeRestorer(
+ radioController = radioController,
+ serviceRepository = serviceRepository,
+ discoveryDao = discoveryDao,
+ applicationScope = applicationScope(backgroundScope),
+ meshPrefs = meshPrefs,
+ )
+
+ val result =
+ restorer.schedule(
+ DiscoveryHomeRestorePlan(
+ sessionId = sessionId,
+ deviceAddress = device,
+ loraConfig = loraConfig,
+ primaryChannel = null,
+ restorePrimaryChannel = false,
+ finalStatus = DiscoverySessionStatus.FAILED,
+ ),
+ )
+ var retryWindowMs = 0L
+ var retryDelayMs = DiscoveryHomeRestorer.RETRY_DELAY_MS
+ repeat(DiscoveryHomeRestorer.MAX_RESTORE_ATTEMPTS - 1) {
+ retryWindowMs += retryDelayMs
+ retryDelayMs =
+ (retryDelayMs * DiscoveryHomeRestorer.RETRY_BACKOFF_MULTIPLIER).coerceAtMost(
+ DiscoveryHomeRestorer.MAX_RETRY_DELAY_MS,
+ )
+ }
+ advanceTimeBy(retryWindowMs)
+ runCurrent()
+
+ assertFalse(result.await())
+ assertEquals(DiscoveryHomeRestorer.MAX_RESTORE_ATTEMPTS, radioController.localConfigWriteAttempts)
+ assertEquals(DiscoverySessionStatus.UNRESTORABLE, discoveryDao.getSession(sessionId)?.completionStatus)
+ assertFalse(restorer.hasUnsatisfiedRestoreFor(device))
+ assertTrue(
+ restorer.awaitBeforeScan(device),
+ "a durably unrestorable restore must not leave a process-lifetime same-device admission barrier",
+ )
+ }
+
+ @Test
+ fun unrestorablePersistenceFailureKeepsSameDeviceBarrierRecoverable() = runTest {
+ val device = "x:UNRESTORABLE-PERSISTENCE"
+ val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress(device) }
+ val serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Connected) }
+ val radioController =
+ DiscoveryTestRadioController().apply {
+ selectedDeviceAddress = device
+ failLocalConfigWritesRemaining = Int.MAX_VALUE
+ }
+ val backingDao = SharedInMemoryDiscoveryDao()
+ val loraConfig = Config.LoRaConfig(use_preset = true)
+ val sessionId =
+ backingDao.insertSession(
+ DiscoverySessionEntity(
+ timestamp = 1L,
+ presetsScanned = "LONG_FAST",
+ homePreset = "LONG_FAST",
+ completionStatus = DiscoverySessionStatus.RESTORE_PENDING_FAILED,
+ deviceAddress = device,
+ homeLoraConfig = loraConfig,
+ ),
+ )
+ val discoveryDao =
+ object : DiscoveryDao by backingDao {
+ override suspend fun updateRecoverableSessionCompletionStatus(sessionId: Long, status: String): Int = 0
+ }
+ val restorer =
+ DiscoveryHomeRestorer(
+ radioController = radioController,
+ serviceRepository = serviceRepository,
+ discoveryDao = discoveryDao,
+ applicationScope = applicationScope(backgroundScope),
+ meshPrefs = meshPrefs,
+ )
+
+ val result =
+ restorer.schedule(
+ DiscoveryHomeRestorePlan(
+ sessionId = sessionId,
+ deviceAddress = device,
+ loraConfig = loraConfig,
+ primaryChannel = null,
+ restorePrimaryChannel = false,
+ finalStatus = DiscoverySessionStatus.FAILED,
+ ),
+ )
+ var retryWindowMs = 0L
+ var retryDelayMs = DiscoveryHomeRestorer.RETRY_DELAY_MS
+ repeat(DiscoveryHomeRestorer.MAX_RESTORE_ATTEMPTS - 1) {
+ retryWindowMs += retryDelayMs
+ retryDelayMs =
+ (retryDelayMs * DiscoveryHomeRestorer.RETRY_BACKOFF_MULTIPLIER).coerceAtMost(
+ DiscoveryHomeRestorer.MAX_RETRY_DELAY_MS,
+ )
+ }
+ advanceTimeBy(retryWindowMs)
+ runCurrent()
+
+ assertFalse(result.await())
+ assertEquals(DiscoverySessionStatus.RESTORE_PENDING_FAILED, backingDao.getSession(sessionId)?.completionStatus)
+ assertTrue(restorer.hasUnsatisfiedRestoreFor(device))
+ assertFalse(
+ restorer.awaitBeforeScan(device),
+ "a failed terminal-status write must keep the recoverable same-device barrier",
+ )
+ }
+}
diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryMapFilterTest.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryMapFilterTest.kt
index 031c041a43..7290fec1fb 100644
--- a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryMapFilterTest.kt
+++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryMapFilterTest.kt
@@ -18,14 +18,14 @@
package org.meshtastic.feature.discovery
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.CoroutineStart
+import kotlinx.coroutines.async
+import kotlinx.coroutines.flow.first
import kotlinx.coroutines.test.runTest
-import org.meshtastic.core.database.dao.DiscoveryDao
import org.meshtastic.core.database.entity.DiscoveredNodeEntity
import org.meshtastic.core.database.entity.DiscoveryPresetResultEntity
import org.meshtastic.core.database.entity.DiscoverySessionEntity
+import org.meshtastic.core.database.entity.DiscoverySessionStatus
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFalse
@@ -112,20 +112,22 @@ class DiscoveryMapFilterTest {
// region Preset results loaded
@Test
- fun presetResults_loadedFromDao() = runTest {
- val dao = MapTestDao()
+ fun sharedDaoFlowsObserveWritesAfterSubscription() = runTest {
+ val dao = SharedInMemoryDiscoveryDao()
val sessionId = dao.insertSession(testSession())
- dao.insertPresetResult(DiscoveryPresetResultEntity(sessionId = sessionId, presetName = "LONG_FAST"))
+ val observedResults =
+ async(start = CoroutineStart.UNDISPATCHED) { dao.getPresetResultsFlow(sessionId).first { it.size == 2 } }
+ val presetResultId =
+ dao.insertPresetResult(DiscoveryPresetResultEntity(sessionId = sessionId, presetName = "LONG_FAST"))
dao.insertPresetResult(DiscoveryPresetResultEntity(sessionId = sessionId, presetName = "SHORT_FAST"))
+ val observedNodes =
+ async(start = CoroutineStart.UNDISPATCHED) {
+ dao.getDiscoveredNodesFlow(presetResultId).first { it.isNotEmpty() }
+ }
+ dao.insertDiscoveredNode(DiscoveredNodeEntity(presetResultId = presetResultId, nodeNum = 123L))
- val vm = DiscoveryMapViewModel(sessionId = sessionId, discoveryDao = dao)
- // safeLaunch runs in UnconfinedTestDispatcher-like context within the VM
- // Access the loaded state
- val results = vm.presetResults.value
- // The VM loads asynchronously, so results may still be loading.
- // Verify the DAO has the right data at minimum.
- val daoResults = dao.getPresetResults(sessionId)
- assertEquals(2, daoResults.size)
+ assertEquals(setOf("LONG_FAST", "SHORT_FAST"), observedResults.await().map { it.presetName }.toSet())
+ assertEquals(listOf(123L), observedNodes.await().map { it.nodeNum })
}
// endregion
@@ -133,7 +135,7 @@ class DiscoveryMapFilterTest {
// region Helpers
private fun createViewModel(): DiscoveryMapViewModel {
- val dao = MapTestDao()
+ val dao = SharedInMemoryDiscoveryDao()
return DiscoveryMapViewModel(sessionId = 1L, discoveryDao = dao)
}
@@ -141,114 +143,8 @@ class DiscoveryMapFilterTest {
timestamp = 1_000_000L,
presetsScanned = "LONG_FAST",
homePreset = "LONG_FAST",
- completionStatus = "complete",
+ completionStatus = DiscoverySessionStatus.COMPLETE,
)
// endregion
}
-
-// region In-memory DAO for map filter tests
-
-private class MapTestDao : DiscoveryDao {
- private var nextSessionId = 1L
- private var nextPresetResultId = 1L
- private var nextNodeId = 1L
-
- private val sessions = mutableMapOf<Long, DiscoverySessionEntity>()
- private val presetResults = mutableMapOf<Long, DiscoveryPresetResultEntity>()
- private val discoveredNodes = mutableMapOf<Long, DiscoveredNodeEntity>()
-
- override suspend fun insertSession(session: DiscoverySessionEntity): Long {
- val id = nextSessionId++
- sessions[id] = session.copy(id = id)
- return id
- }
-
- override suspend fun updateSession(session: DiscoverySessionEntity) {
- sessions[session.id] = session
- }
-
- override fun getAllSessions(): Flow<List<DiscoverySessionEntity>> =
- flowOf(sessions.values.sortedByDescending { it.timestamp })
-
- override suspend fun getAllSessionsSnapshot(): List<DiscoverySessionEntity> = sessions.values.toList()
-
- override suspend fun getSession(sessionId: Long) = sessions[sessionId]
-
- override fun getSessionFlow(sessionId: Long): Flow<DiscoverySessionEntity?> = MutableStateFlow(sessions[sessionId])
-
- override suspend fun deleteSession(sessionId: Long) {
- sessions.remove(sessionId)
- val resultIds = presetResults.values.filter { it.sessionId == sessionId }.map { it.id }
- resultIds.forEach { rid ->
- discoveredNodes.entries.removeAll { it.value.presetResultId == rid }
- presetResults.remove(rid)
- }
- }
-
- override suspend fun insertPresetResult(result: DiscoveryPresetResultEntity): Long {
- val id = nextPresetResultId++
- presetResults[id] = result.copy(id = id)
- return id
- }
-
- override suspend fun updatePresetResult(result: DiscoveryPresetResultEntity) {
- presetResults[result.id] = result
- }
-
- override suspend fun getPresetResults(sessionId: Long) = presetResults.values.filter { it.sessionId == sessionId }
-
- override fun getPresetResultsFlow(sessionId: Long) =
- flowOf(presetResults.values.filter { it.sessionId == sessionId })
-
- override suspend fun insertDiscoveredNode(node: DiscoveredNodeEntity): Long {
- val id = nextNodeId++
- discoveredNodes[id] = node.copy(id = id)
- return id
- }
-
- override suspend fun insertDiscoveredNodes(nodes: List<DiscoveredNodeEntity>) {
- nodes.forEach { insertDiscoveredNode(it) }
- }
-
- override suspend fun updateDiscoveredNode(node: DiscoveredNodeEntity) {
- discoveredNodes[node.id] = node
- }
-
- override suspend fun getDiscoveredNodes(presetResultId: Long) =
- discoveredNodes.values.filter { it.presetResultId == presetResultId }
-
- override fun getDiscoveredNodesFlow(presetResultId: Long) =
- flowOf(discoveredNodes.values.filter { it.presetResultId == presetResultId })
-
- override suspend fun getUniqueNodeNums(sessionId: Long) = presetResults.values
- .filter { it.sessionId == sessionId }
- .flatMap { pr -> discoveredNodes.values.filter { it.presetResultId == pr.id } }
- .map { it.nodeNum }
- .distinct()
-
- override suspend fun getUniqueNodeCount(sessionId: Long) = getUniqueNodeNums(sessionId).size
-
- override suspend fun getMaxDistance(sessionId: Long) = presetResults.values
- .filter { it.sessionId == sessionId }
- .flatMap { pr -> discoveredNodes.values.filter { it.presetResultId == pr.id } }
- .mapNotNull { it.distanceFromUser }
- .maxOrNull()
-
- override suspend fun getSessionWithResults(sessionId: Long) = sessions[sessionId]
-
- override suspend fun markInterruptedSessions() {
- sessions.keys.toList().forEach { key ->
- val session = sessions[key]!!
- if (session.completionStatus == "in_progress") {
- sessions[key] = session.copy(completionStatus = "interrupted")
- }
- }
- }
-
- override suspend fun getInterruptedSession(deviceAddress: String): DiscoverySessionEntity? = sessions.values
- .filter { it.deviceAddress == deviceAddress && it.completionStatus in setOf("in_progress", "interrupted") }
- .maxByOrNull { it.timestamp }
-}
-
-// endregion
diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryPacketCollectionTest.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryPacketCollectionTest.kt
index e424c55430..ea9092755e 100644
--- a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryPacketCollectionTest.kt
+++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryPacketCollectionTest.kt
@@ -20,17 +20,12 @@ package org.meshtastic.feature.discovery
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.delay
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.test.TestScope
import kotlinx.coroutines.test.UnconfinedTestDispatcher
import kotlinx.coroutines.test.runTest
import okio.ByteString
import okio.ByteString.Companion.toByteString
import org.meshtastic.core.common.di.ApplicationCoroutineScope
-import org.meshtastic.core.database.dao.DiscoveryDao
-import org.meshtastic.core.database.entity.DiscoveredNodeEntity
import org.meshtastic.core.database.entity.DiscoveryPresetResultEntity
import org.meshtastic.core.database.entity.DiscoverySessionEntity
import org.meshtastic.core.di.CoroutineDispatchers
@@ -66,7 +61,7 @@ import kotlin.test.assertTrue
*/
class DiscoveryPacketCollectionTest {
- private val radioController = FakeRadioController()
+ private val radioController = FakeRadioController().apply { selectedDeviceAddress = DEFAULT_DEVICE_ADDRESS }
private val serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Connected) }
private val nodeRepository = FakeNodeRepository()
private val radioConfigRepository =
@@ -78,9 +73,9 @@ class DiscoveryPacketCollectionTest {
)
}
private val collectorRegistry = PacketTestCollectorRegistry()
- private val discoveryDao = InMemoryDiscoveryDao()
+ private val discoveryDao = SharedInMemoryDiscoveryDao()
private val aiProvider = PacketTestAiProvider()
- private val meshPrefs = FakeMeshPrefs()
+ private val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress(DEFAULT_DEVICE_ADDRESS) }
private fun createEngine(testScope: TestScope): DiscoveryScanEngine {
val testDispatcher = UnconfinedTestDispatcher(testScope.testScheduler)
@@ -314,6 +309,10 @@ class DiscoveryPacketCollectionTest {
)
// endregion
+
+ private companion object {
+ const val DEFAULT_DEVICE_ADDRESS = "x:TEST-DEVICE"
+ }
}
// region Inline test doubles
@@ -332,105 +331,3 @@ private class PacketTestAiProvider : DiscoverySummaryAiProvider {
override suspend fun generatePresetSummary(result: DiscoveryPresetResultEntity): String? = null
}
-
-private class InMemoryDiscoveryDao : DiscoveryDao {
- private var nextSessionId = 1L
- private var nextPresetResultId = 1L
- private var nextNodeId = 1L
-
- val sessions = mutableMapOf<Long, DiscoverySessionEntity>()
- val presetResults = mutableMapOf<Long, DiscoveryPresetResultEntity>()
- val discoveredNodes = mutableMapOf<Long, DiscoveredNodeEntity>()
-
- override suspend fun insertSession(session: DiscoverySessionEntity): Long {
- val id = nextSessionId++
- sessions[id] = session.copy(id = id)
- return id
- }
-
- override suspend fun updateSession(session: DiscoverySessionEntity) {
- sessions[session.id] = session
- }
-
- override fun getAllSessions(): Flow<List<DiscoverySessionEntity>> =
- flowOf(sessions.values.sortedByDescending { it.timestamp })
-
- override suspend fun getAllSessionsSnapshot(): List<DiscoverySessionEntity> = sessions.values.toList()
-
- override suspend fun getSession(sessionId: Long): DiscoverySessionEntity? = sessions[sessionId]
-
- override fun getSessionFlow(sessionId: Long): Flow<DiscoverySessionEntity?> = MutableStateFlow(sessions[sessionId])
-
- override suspend fun deleteSession(sessionId: Long) {
- sessions.remove(sessionId)
- val resultIds = presetResults.values.filter { it.sessionId == sessionId }.map { it.id }
- resultIds.forEach { rid ->
- discoveredNodes.entries.removeAll { it.value.presetResultId == rid }
- presetResults.remove(rid)
- }
- }
-
- override suspend fun insertPresetResult(result: DiscoveryPresetResultEntity): Long {
- val id = nextPresetResultId++
- presetResults[id] = result.copy(id = id)
- return id
- }
-
- override suspend fun updatePresetResult(result: DiscoveryPresetResultEntity) {
- presetResults[result.id] = result
- }
-
- override suspend fun getPresetResults(sessionId: Long) = presetResults.values.filter { it.sessionId == sessionId }
-
- override fun getPresetResultsFlow(sessionId: Long) =
- flowOf(presetResults.values.filter { it.sessionId == sessionId })
-
- override suspend fun insertDiscoveredNode(node: DiscoveredNodeEntity): Long {
- val id = nextNodeId++
- discoveredNodes[id] = node.copy(id = id)
- return id
- }
-
- override suspend fun insertDiscoveredNodes(nodes: List<DiscoveredNodeEntity>) {
- nodes.forEach { insertDiscoveredNode(it) }
- }
-
- override suspend fun updateDiscoveredNode(node: DiscoveredNodeEntity) {
- discoveredNodes[node.id] = node
- }
-
- override suspend fun getDiscoveredNodes(presetResultId: Long) =
- discoveredNodes.values.filter { it.presetResultId == presetResultId }
-
- override fun getDiscoveredNodesFlow(presetResultId: Long) =
- flowOf(discoveredNodes.values.filter { it.presetResultId == presetResultId })
-
- override suspend fun getUniqueNodeNums(sessionId: Long) = presetResults.values
- .filter { it.sessionId == sessionId }
- .flatMap { pr -> discoveredNodes.values.filter { it.presetResultId == pr.id } }
- .map { it.nodeNum }
- .distinct()
-
- override suspend fun getUniqueNodeCount(sessionId: Long) = getUniqueNodeNums(sessionId).size
-
- override suspend fun getMaxDistance(sessionId: Long) = presetResults.values
- .filter { it.sessionId == sessionId }
- .flatMap { pr -> discoveredNodes.values.filter { it.presetResultId == pr.id } }
- .mapNotNull { it.distanceFromUser }
- .maxOrNull()
-
- override suspend fun getSessionWithResults(sessionId: Long) = sessions[sessionId]
-
- override suspend fun markInterruptedSessions() {
- sessions.keys.toList().forEach { key ->
- val session = sessions[key]!!
- if (session.completionStatus == "in_progress") {
- sessions[key] = session.copy(completionStatus = "interrupted")
- }
- }
- }
-
- override suspend fun getInterruptedSession(deviceAddress: String): DiscoverySessionEntity? = sessions.values
- .filter { it.deviceAddress == deviceAddress && it.completionStatus in setOf("in_progress", "interrupted") }
- .maxByOrNull { it.timestamp }
-}
diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngineTest.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngineTest.kt
index 53e45eed1f..1785f44731 100644
--- a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngineTest.kt
+++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryScanEngineTest.kt
@@ -18,14 +18,14 @@
package org.meshtastic.feature.discovery
+import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.async
import kotlinx.coroutines.delay
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.launch
import kotlinx.coroutines.test.TestScope
import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.advanceTimeBy
import kotlinx.coroutines.test.advanceUntilIdle
import kotlinx.coroutines.test.runCurrent
import kotlinx.coroutines.test.runTest
@@ -36,6 +36,7 @@ import org.meshtastic.core.database.dao.DiscoveryDao
import org.meshtastic.core.database.entity.DiscoveredNodeEntity
import org.meshtastic.core.database.entity.DiscoveryPresetResultEntity
import org.meshtastic.core.database.entity.DiscoverySessionEntity
+import org.meshtastic.core.database.entity.DiscoverySessionStatus
import org.meshtastic.core.di.CoroutineDispatchers
import org.meshtastic.core.model.ChannelOption
import org.meshtastic.core.model.ConnectionState
@@ -50,7 +51,6 @@ import org.meshtastic.core.repository.PacketQueueRejectedException
import org.meshtastic.core.testing.FakeMeshPrefs
import org.meshtastic.core.testing.FakeNodeRepository
import org.meshtastic.core.testing.FakeRadioConfigRepository
-import org.meshtastic.core.testing.FakeRadioController
import org.meshtastic.core.testing.FakeServiceRepository
import org.meshtastic.feature.discovery.ai.DiscoverySummaryAiProvider
import org.meshtastic.proto.Config
@@ -71,111 +71,93 @@ import kotlin.test.assertTrue
// region Inline fakes
-/** In-memory fake of [DiscoveryDao] for unit tests. */
-private class FakeDiscoveryDao : DiscoveryDao {
- private var nextSessionId = 1L
- private var nextPresetResultId = 1L
- private var nextNodeId = 1L
+/** [DiscoveryDao] wrapper that adds deterministic synchronization and failure hooks for scan-engine tests. */
+private class FakeDiscoveryDao(private val delegate: SharedInMemoryDiscoveryDao = SharedInMemoryDiscoveryDao()) :
+ DiscoveryDao by delegate {
+ val sessions: Map<Long, DiscoverySessionEntity>
+ get() = delegate.sessions
- val sessions = mutableMapOf<Long, DiscoverySessionEntity>()
- val presetResults = mutableMapOf<Long, DiscoveryPresetResultEntity>()
- val discoveredNodes = mutableMapOf<Long, DiscoveredNodeEntity>()
+ val presetResults: Map<Long, DiscoveryPresetResultEntity>
+ get() = delegate.presetResults
- override suspend fun insertSession(session: DiscoverySessionEntity): Long {
- val id = nextSessionId++
- sessions[id] = session.copy(id = id)
- return id
- }
+ val discoveredNodes: Map<Long, DiscoveredNodeEntity>
+ get() = delegate.discoveredNodes
- override suspend fun updateSession(session: DiscoverySessionEntity) {
- sessions[session.id] = session
- }
+ var nextInsertSessionEntered: CompletableDeferred<Unit>? = null
+ var releaseNextInsertSession: CompletableDeferred<Unit>? = null
+ var nextSessionWriteEntered: CompletableDeferred<Unit>? = null
+ var releaseNextSessionWrite: CompletableDeferred<Unit>? = null
+ var nextInsertPresetResultEntered: CompletableDeferred<Unit>? = null
+ var releaseNextInsertPresetResult: CompletableDeferred<Unit>? = null
+ var nextSessionWriteFailure: Exception? = null
- override fun getAllSessions(): Flow<List<DiscoverySessionEntity>> =
- flowOf(sessions.values.sortedByDescending { it.timestamp })
+ suspend fun seedSession(session: DiscoverySessionEntity) = delegate.seedSession(session)
- override suspend fun getAllSessionsSnapshot(): List<DiscoverySessionEntity> = sessions.values.toList()
-
- override suspend fun getSession(sessionId: Long): DiscoverySessionEntity? = sessions[sessionId]
-
- override fun getSessionFlow(sessionId: Long): Flow<DiscoverySessionEntity?> = MutableStateFlow(sessions[sessionId])
-
- override suspend fun deleteSession(sessionId: Long) {
- sessions.remove(sessionId)
- val resultIds = presetResults.values.filter { it.sessionId == sessionId }.map { it.id }
- resultIds.forEach { resultId ->
- discoveredNodes.entries.removeAll { it.value.presetResultId == resultId }
- presetResults.remove(resultId)
+ override suspend fun insertSession(session: DiscoverySessionEntity): Long {
+ nextInsertSessionEntered?.also { entered ->
+ nextInsertSessionEntered = null
+ entered.complete(Unit)
}
+ releaseNextInsertSession?.also { release ->
+ releaseNextInsertSession = null
+ release.await()
+ }
+ return delegate.insertSession(session)
}
- override suspend fun insertPresetResult(result: DiscoveryPresetResultEntity): Long {
- val id = nextPresetResultId++
- presetResults[id] = result.copy(id = id)
- return id
- }
-
- override suspend fun updatePresetResult(result: DiscoveryPresetResultEntity) {
- presetResults[result.id] = result
- }
-
- override suspend fun getPresetResults(sessionId: Long): List<DiscoveryPresetResultEntity> =
- presetResults.values.filter { it.sessionId == sessionId }
-
- override fun getPresetResultsFlow(sessionId: Long): Flow<List<DiscoveryPresetResultEntity>> =
- flowOf(getPresetResultsSynchronous(sessionId))
-
- private fun getPresetResultsSynchronous(sessionId: Long): List<DiscoveryPresetResultEntity> =
- presetResults.values.filter { it.sessionId == sessionId }
-
- override suspend fun insertDiscoveredNode(node: DiscoveredNodeEntity): Long {
- val id = nextNodeId++
- discoveredNodes[id] = node.copy(id = id)
- return id
+ private suspend fun applySessionWriteHooks() {
+ nextSessionWriteEntered?.also { entered ->
+ nextSessionWriteEntered = null
+ entered.complete(Unit)
+ }
+ releaseNextSessionWrite?.also { release ->
+ releaseNextSessionWrite = null
+ release.await()
+ }
+ nextSessionWriteFailure?.also { failure ->
+ nextSessionWriteFailure = null
+ throw failure
+ }
}
- override suspend fun insertDiscoveredNodes(nodes: List<DiscoveredNodeEntity>) {
- nodes.forEach { insertDiscoveredNode(it) }
+ override suspend fun updateSession(session: DiscoverySessionEntity) {
+ applySessionWriteHooks()
+ delegate.updateSession(session)
}
- override suspend fun updateDiscoveredNode(node: DiscoveredNodeEntity) {
- discoveredNodes[node.id] = node
+ override suspend fun updateSessionAggregates(
+ sessionId: Long,
+ totalUniqueNodes: Int,
+ totalDwellSeconds: Long,
+ totalMessages: Int,
+ totalSensorPackets: Int,
+ furthestNodeDistance: Double,
+ avgChannelUtilization: Double,
+ ) {
+ applySessionWriteHooks()
+ delegate.updateSessionAggregates(
+ sessionId = sessionId,
+ totalUniqueNodes = totalUniqueNodes,
+ totalDwellSeconds = totalDwellSeconds,
+ totalMessages = totalMessages,
+ totalSensorPackets = totalSensorPackets,
+ furthestNodeDistance = furthestNodeDistance,
+ avgChannelUtilization = avgChannelUtilization,
+ )
}
- override suspend fun getDiscoveredNodes(presetResultId: Long): List<DiscoveredNodeEntity> =
- discoveredNodes.values.filter { it.presetResultId == presetResultId }
-
- override fun getDiscoveredNodesFlow(presetResultId: Long): Flow<List<DiscoveredNodeEntity>> =
- flowOf(discoveredNodes.values.filter { it.presetResultId == presetResultId })
-
- override suspend fun getUniqueNodeNums(sessionId: Long): List<Long> = presetResults.values
- .filter { it.sessionId == sessionId }
- .flatMap { pr -> discoveredNodes.values.filter { it.presetResultId == pr.id } }
- .map { it.nodeNum }
- .distinct()
-
- override suspend fun getUniqueNodeCount(sessionId: Long): Int = getUniqueNodeNums(sessionId).size
-
- override suspend fun getMaxDistance(sessionId: Long): Double? = presetResults.values
- .filter { it.sessionId == sessionId }
- .flatMap { pr -> discoveredNodes.values.filter { it.presetResultId == pr.id } }
- .mapNotNull { it.distanceFromUser }
- .maxOrNull()
-
- override suspend fun getSessionWithResults(sessionId: Long): DiscoverySessionEntity? = sessions[sessionId]
-
- override suspend fun markInterruptedSessions() {
- sessions.keys.toList().forEach { key ->
- val session = sessions[key]!!
- if (session.completionStatus == "in_progress") {
- sessions[key] = session.copy(completionStatus = "interrupted")
- }
+ override suspend fun insertPresetResult(result: DiscoveryPresetResultEntity): Long {
+ val id = delegate.insertPresetResult(result)
+ nextInsertPresetResultEntered?.also { entered ->
+ nextInsertPresetResultEntered = null
+ entered.complete(Unit)
}
+ releaseNextInsertPresetResult?.also { release ->
+ releaseNextInsertPresetResult = null
+ release.await()
+ }
+ return id
}
-
- override suspend fun getInterruptedSession(deviceAddress: String): DiscoverySessionEntity? = sessions.values
- .filter { it.deviceAddress == deviceAddress && it.completionStatus in setOf("in_progress", "interrupted") }
- .maxByOrNull { it.timestamp }
}
/** Simple fake collector registry that tracks registration. */
@@ -199,7 +181,8 @@ private class FakeAiProvider : DiscoverySummaryAiProvider {
class DiscoveryScanEngineTest {
- private val radioController = FakeRadioController()
+ private val radioController =
+ DiscoveryTestRadioController().apply { selectedDeviceAddress = DEFAULT_DEVICE_ADDRESS }
private val serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Connected) }
private val nodeRepository = FakeNodeRepository()
private val radioConfigRepository =
@@ -213,7 +196,7 @@ class DiscoveryScanEngineTest {
private val collectorRegistry = FakeCollectorRegistry()
private val discoveryDao = FakeDiscoveryDao()
private val aiProvider = FakeAiProvider()
- private val meshPrefs = FakeMeshPrefs()
+ private val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress(DEFAULT_DEVICE_ADDRESS) }
/** Creates a [DiscoveryScanEngine] wired to test dispatchers sharing the given [testScope]'s scheduler. */
private fun createEngine(testScope: TestScope): DiscoveryScanEngine {
@@ -239,6 +222,11 @@ class DiscoveryScanEngineTest {
private val testPresets = listOf(ChannelOption.LONG_FAST)
+ private fun selectDevice(address: String?) {
+ meshPrefs.setDeviceAddress(address)
+ radioController.selectedDeviceAddress = address
+ }
+
/**
* After [DiscoveryScanEngine.startScan], the state is set to [DiscoveryScanState.Shifting] synchronously. This
* helper asserts that the engine is active — no real-time wait needed.
@@ -319,7 +307,7 @@ class DiscoveryScanEngineTest {
// Session should be persisted (happens synchronously inside startScan)
assertEquals(1, discoveryDao.sessions.size)
val session = discoveryDao.sessions.values.first()
- assertEquals("in_progress", session.completionStatus)
+ assertEquals(DiscoverySessionStatus.IN_PROGRESS, session.completionStatus)
assertEquals("LONG_FAST", session.presetsScanned)
assertEquals("LONG_FAST", session.homePreset)
@@ -337,6 +325,58 @@ class DiscoveryScanEngineTest {
engine.stopScan()
}
+ @Test
+ fun deviceSwitchDuringSessionInsertAbortsScanBeforePublication() = runTest {
+ val firstDevice = "x:FIRST"
+ selectDevice(firstDevice)
+ val insertEntered = CompletableDeferred<Unit>()
+ val releaseInsert = CompletableDeferred<Unit>()
+ discoveryDao.nextInsertSessionEntered = insertEntered
+ discoveryDao.releaseNextInsertSession = releaseInsert
+ val engine = createEngine(this)
+
+ val startScan = async { engine.startScan(testPresets, dwellDurationSeconds = 10) }
+ insertEntered.await()
+ selectDevice("x:SECOND")
+ releaseInsert.complete(Unit)
+ startScan.await()
+
+ assertEquals(
+ DiscoveryScanState.Failed("Selected radio changed while preparing the scan"),
+ engine.scanState.value,
+ )
+ assertTrue(discoveryDao.sessions.isEmpty(), "Aborted preparation must not leave an in-progress session")
+ assertNull(engine.currentSession.value)
+ assertNull(collectorRegistry.collector)
+ assertTrue(radioController.configWrites.isEmpty())
+ }
+
+ @Test
+ fun transportRestartDuringSessionInsertAbortsScanBeforePublication() = runTest {
+ selectDevice("x:CURRENT")
+ radioController.setSessionGeneration(7L)
+ val insertEntered = CompletableDeferred<Unit>()
+ val releaseInsert = CompletableDeferred<Unit>()
+ discoveryDao.nextInsertSessionEntered = insertEntered
+ discoveryDao.releaseNextInsertSession = releaseInsert
+ val engine = createEngine(this)
+
+ val startScan = async { engine.startScan(testPresets, dwellDurationSeconds = 10) }
+ insertEntered.await()
+ radioController.setSessionGeneration(8L)
+ releaseInsert.complete(Unit)
+ startScan.await()
+
+ assertEquals(
+ DiscoveryScanState.Failed("Selected radio changed while preparing the scan"),
+ engine.scanState.value,
+ )
+ assertTrue(discoveryDao.sessions.isEmpty(), "A replaced transport must not publish a prepared session")
+ assertNull(engine.currentSession.value)
+ assertNull(collectorRegistry.collector)
+ assertTrue(radioController.configWrites.isEmpty(), "A replaced transport must not be retuned")
+ }
+
@Test
fun stopScanPersistsResultsAndTransitionsToIdle() = runTest {
val engine = createEngine(this)
@@ -357,9 +397,12 @@ class DiscoveryScanEngineTest {
// Collector should be unregistered
assertNull(collectorRegistry.collector)
- // Session should be finalized with "stopped" status
+ // stopScan returns after process-scope restoration ownership is established, not after its settle delay.
val session = discoveryDao.sessions.values.first()
- assertEquals("stopped", session.completionStatus)
+ assertEquals(DiscoverySessionStatus.RESTORE_PENDING_STOPPED, session.completionStatus)
+ advanceTimeBy(DiscoveryHomeRestorer.POST_RESTORE_SETTLE_DELAY_MS)
+ runCurrent()
+ assertEquals(DiscoverySessionStatus.STOPPED, discoveryDao.sessions.values.first().completionStatus)
}
@Test
@@ -369,7 +412,7 @@ class DiscoveryScanEngineTest {
// Immediately after startScan, the session should exist with "in_progress"
val session = discoveryDao.sessions.values.first()
- assertEquals("in_progress", session.completionStatus)
+ assertEquals(DiscoverySessionStatus.IN_PROGRESS, session.completionStatus)
// Wait for the scan loop to start, then verify active
assertScanActive(engine)
@@ -650,11 +693,39 @@ class DiscoveryScanEngineTest {
assertFalse(engine.isActive)
assertNull(collectorRegistry.collector, "collector should be unregistered")
- assertEquals("failed", discoveryDao.sessions.values.first().completionStatus)
- // The last LoRa config applied is the home preset (LONG_FAST), not the scan preset (SHORT_FAST).
+ assertEquals(
+ DiscoverySessionStatus.RESTORE_PENDING_FAILED,
+ discoveryDao.sessions.values.first().completionStatus,
+ )
+ assertEquals(ChannelOption.SHORT_FAST.modemPreset, radioController.lastLocalConfig?.lora?.modem_preset)
+
+ serviceRepository.setConnectionState(ConnectionState.Connected)
+ advanceUntilIdle()
+ assertEquals(DiscoverySessionStatus.FAILED, discoveryDao.sessions.values.first().completionStatus)
assertEquals(ChannelOption.LONG_FAST.modemPreset, radioController.lastLocalConfig?.lora?.modem_preset)
}
+ @Test
+ fun dwellFailurePersistsNodesCollectedBeforeTheAbort() = runTest {
+ val engine = createEngine(this)
+ engine.startScan(listOf(ChannelOption.SHORT_FAST), dwellDurationSeconds = 60)
+ while (engine.scanState.value !is DiscoveryScanState.Dwell) {
+ delay(100)
+ }
+ engine.onPacketReceived(
+ createPositionMeshPacket(from = 54321, latI = 0, lonI = 0),
+ createDataPacket(from = 54321),
+ )
+
+ serviceRepository.setConnectionState(ConnectionState.Disconnected)
+ advanceUntilIdle()
+
+ val result = discoveryDao.presetResults.values.single()
+ assertEquals(ChannelOption.SHORT_FAST.name, result.presetName)
+ assertEquals(1, result.uniqueNodes)
+ assertEquals(1, discoveryDao.discoveredNodes.size)
+ }
+
@Test
fun stopScanRestoresHomePreset() = runTest {
val engine = createEngine(this)
@@ -667,6 +738,292 @@ class DiscoveryScanEngineTest {
assertEquals(ChannelOption.LONG_FAST.modemPreset, radioController.lastLocalConfig?.lora?.modem_preset)
}
+ @Test
+ fun concurrentStopsShareOneTerminalCleanupAndHomeRestore() = runTest {
+ val engine = createEngine(this)
+ engine.startScan(listOf(ChannelOption.SHORT_FAST), dwellDurationSeconds = 60)
+ assertScanActive(engine)
+ val originalSessionId = checkNotNull(engine.currentSession.value).id
+ val updateEntered = CompletableDeferred<Unit>()
+ val releaseUpdate = CompletableDeferred<Unit>()
+ discoveryDao.nextSessionWriteEntered = updateEntered
+ discoveryDao.releaseNextSessionWrite = releaseUpdate
+
+ val firstStop = async { engine.stopScan() }
+ updateEntered.await()
+ val redundantStop = async { engine.stopScan() }
+ runCurrent()
+
+ assertFalse(redundantStop.isCompleted, "all stop callers must await the elected terminal cleanup")
+ releaseUpdate.complete(Unit)
+ firstStop.await()
+ redundantStop.await()
+ advanceUntilIdle()
+
+ val homeWrites =
+ radioController.configWrites.count { it.lora?.modem_preset == ChannelOption.LONG_FAST.modemPreset }
+ assertEquals(1, homeWrites)
+ assertEquals(1, discoveryDao.presetResults.size, "joined terminal cleanup must persist the dwell exactly once")
+ assertEquals(setOf(originalSessionId), discoveryDao.sessions.keys)
+ }
+
+ @Test
+ fun stopRacingLastDwellPersistenceDoesNotInsertTheResultTwice() = runTest {
+ val insertEntered = CompletableDeferred<Unit>()
+ val releaseInsert = CompletableDeferred<Unit>()
+ discoveryDao.nextInsertPresetResultEntered = insertEntered
+ discoveryDao.releaseNextInsertPresetResult = releaseInsert
+ val engine = createEngine(this)
+
+ engine.startScan(listOf(ChannelOption.SHORT_FAST), dwellDurationSeconds = 1)
+ advanceTimeBy(1_001L)
+ insertEntered.await()
+ val stop = async { engine.stopScan() }
+ runCurrent()
+ assertFalse(stop.isCompleted, "stop must serialize with the final dwell persistence")
+
+ releaseInsert.complete(Unit)
+ stop.await()
+ advanceUntilIdle()
+
+ assertEquals(1, discoveryDao.presetResults.size)
+ }
+
+ @Test
+ fun stopDuringNaturalTerminalCleanupDoesNotReplaceSuccessfulCompletion() = runTest {
+ val engine = createEngine(this)
+ val updateEntered = CompletableDeferred<Unit>()
+ val releaseUpdate = CompletableDeferred<Unit>()
+ discoveryDao.nextSessionWriteEntered = updateEntered
+ discoveryDao.releaseNextSessionWrite = releaseUpdate
+
+ engine.startScan(listOf(ChannelOption.SHORT_FAST), dwellDurationSeconds = 1)
+ advanceTimeBy(1_001L)
+ updateEntered.await()
+ assertEquals(DiscoveryScanState.Analysis, engine.scanState.value)
+
+ engine.stopScan()
+ releaseUpdate.complete(Unit)
+ advanceUntilIdle()
+
+ val state = engine.scanState.value
+ assertTrue(state is DiscoveryScanState.Complete, "expected Complete, was $state")
+ assertEquals(DiscoveryScanState.CompletionOutcome.Success, (state as DiscoveryScanState.Complete).outcome)
+ assertEquals(DiscoverySessionStatus.COMPLETE, discoveryDao.sessions.values.first().completionStatus)
+ }
+
+ @Test
+ fun terminalAggregateWriteDoesNotOverwriteAConcurrentRestoreStatus() = runTest {
+ val engine = createEngine(this)
+ engine.startScan(listOf(ChannelOption.SHORT_FAST), dwellDurationSeconds = 60)
+ assertScanActive(engine)
+ val sessionId = checkNotNull(engine.currentSession.value).id
+ val updateEntered = CompletableDeferred<Unit>()
+ val releaseUpdate = CompletableDeferred<Unit>()
+ discoveryDao.nextSessionWriteEntered = updateEntered
+ discoveryDao.releaseNextSessionWrite = releaseUpdate
+
+ val stop = async { engine.stopScan() }
+ updateEntered.await()
+ discoveryDao.updateRecoverableSessionCompletionStatus(sessionId, DiscoverySessionStatus.COMPLETE)
+ releaseUpdate.complete(Unit)
+ stop.await()
+
+ assertEquals(
+ DiscoverySessionStatus.COMPLETE,
+ discoveryDao.sessions.getValue(sessionId).completionStatus,
+ "terminal aggregate persistence must preserve a status finalized by a concurrent restore",
+ )
+ }
+
+ @Test
+ fun resetAndNewScanCannotDisplaceActiveTerminalCleanup() = runTest {
+ val engine = createEngine(this)
+ engine.startScan(listOf(ChannelOption.SHORT_FAST), dwellDurationSeconds = 60)
+ assertScanActive(engine)
+ val originalSessionId = assertNotNull(engine.currentSession.value).id
+ val updateEntered = CompletableDeferred<Unit>()
+ val releaseUpdate = CompletableDeferred<Unit>()
+ discoveryDao.nextSessionWriteEntered = updateEntered
+ discoveryDao.releaseNextSessionWrite = releaseUpdate
+
+ val stop = async { engine.stopScan() }
+ updateEntered.await()
+ engine.reset()
+ engine.startScan(listOf(ChannelOption.MEDIUM_FAST), dwellDurationSeconds = 60)
+
+ assertEquals(setOf(originalSessionId), discoveryDao.sessions.keys)
+ assertEquals(
+ DiscoverySessionStatus.IN_PROGRESS,
+ discoveryDao.sessions.getValue(originalSessionId).completionStatus,
+ )
+ assertEquals(DiscoveryScanState.Cancelling, engine.scanState.value)
+ assertEquals(
+ originalSessionId,
+ assertNotNull(engine.currentSession.value, "reset must not discard the active session restore plan").id,
+ )
+ assertFalse(
+ radioController.configWrites.any { it.lora?.modem_preset == ChannelOption.MEDIUM_FAST.modemPreset },
+ "a refused restart must not retune the radio",
+ )
+
+ releaseUpdate.complete(Unit)
+ stop.await()
+ advanceUntilIdle()
+ }
+
+ @Test
+ fun targetShiftFailureFailsScanAndStillRestoresHomePreset() = runTest {
+ radioController.failLocalConfigWritesRemaining = 1
+ val engine = createEngine(this)
+
+ engine.startScan(listOf(ChannelOption.SHORT_FAST), dwellDurationSeconds = 60)
+ advanceUntilIdle()
+
+ val state = engine.scanState.value
+ assertTrue(state is DiscoveryScanState.Complete, "expected Complete, was $state")
+ assertEquals(DiscoveryScanState.CompletionOutcome.Failed, (state as DiscoveryScanState.Complete).outcome)
+ assertEquals(DiscoverySessionStatus.FAILED, discoveryDao.sessions.values.first().completionStatus)
+ assertEquals(ChannelOption.LONG_FAST.modemPreset, radioController.lastLocalConfig?.lora?.modem_preset)
+ assertTrue(
+ discoveryDao.presetResults.isEmpty(),
+ "a shift failure runs no dwell, so it must not insert an empty preset result",
+ )
+ }
+
+ @Test
+ fun stopScanRetriesHomeRestoreAfterTransientWriteFailure() = runTest {
+ val engine = createEngine(this)
+ engine.startScan(listOf(ChannelOption.SHORT_FAST), dwellDurationSeconds = 60)
+ assertScanActive(engine)
+ radioController.failLocalConfigWritesRemaining = 1
+
+ engine.stopScan()
+ advanceUntilIdle()
+
+ assertEquals(0, radioController.failLocalConfigWritesRemaining)
+ assertEquals(ChannelOption.LONG_FAST.modemPreset, radioController.lastLocalConfig?.lora?.modem_preset)
+ }
+
+ @Test
+ fun newScanWaitsForPendingHomeRestoreBeforeRetuning() = runTest {
+ val engine = createEngine(this)
+ engine.startScan(listOf(ChannelOption.SHORT_FAST), dwellDurationSeconds = 60)
+ assertScanActive(engine)
+ radioController.failLocalConfigWritesRemaining = 1
+
+ engine.stopScan()
+ engine.reset()
+ val restart = async { engine.startScan(listOf(ChannelOption.MEDIUM_FAST), dwellDurationSeconds = 60) }
+ runCurrent()
+
+ assertFalse(restart.isCompleted, "a new scan must wait for the prior session's home restore")
+ assertFalse(
+ radioController.configWrites.any { it.lora?.modem_preset == ChannelOption.MEDIUM_FAST.modemPreset },
+ "the new scan must not retune before the prior home restore succeeds",
+ )
+
+ // The home restorer's retry delay after the failed config write, plus the settle delay after the
+ // successful restore. Both must elapse before awaitBeforeScan admits the new scan.
+ advanceTimeBy(DiscoveryHomeRestorer.RETRY_DELAY_MS + DiscoveryHomeRestorer.POST_RESTORE_SETTLE_DELAY_MS + 1)
+ restart.await()
+ runCurrent()
+
+ val appliedPresets = radioController.configWrites.mapNotNull { it.lora?.modem_preset }
+ val homeRestoreIndex = appliedPresets.indexOfFirst { it == ChannelOption.LONG_FAST.modemPreset }
+ val newScanIndex = appliedPresets.indexOfFirst { it == ChannelOption.MEDIUM_FAST.modemPreset }
+ assertTrue(homeRestoreIndex >= 0, "the prior session must restore the home preset")
+ assertTrue(newScanIndex > homeRestoreIndex, "the new scan retune must happen after home restoration")
+
+ engine.stopScan()
+ advanceUntilIdle()
+ }
+
+ @Test
+ fun scanIsRefusedWhenHomeLoraConfigCannotBeCaptured() = runTest {
+ radioConfigRepository.setLocalConfigDirect(LocalConfig())
+ val engine = createEngine(this)
+
+ engine.startScan(listOf(ChannelOption.SHORT_FAST), dwellDurationSeconds = 60)
+
+ assertEquals(DiscoveryScanState.Failed("Home LoRa configuration is not available"), engine.scanState.value)
+ assertTrue(discoveryDao.sessions.isEmpty())
+ assertTrue(radioController.configWrites.isEmpty())
+ }
+
+ @Test
+ fun scanIsRefusedWithoutSelectedDeviceOwnership() = runTest {
+ selectDevice(null)
+ val engine = createEngine(this)
+
+ engine.startScan(listOf(ChannelOption.SHORT_FAST), dwellDurationSeconds = 60)
+
+ assertEquals(DiscoveryScanState.Failed("Selected radio is not available"), engine.scanState.value)
+ assertTrue(discoveryDao.sessions.isEmpty())
+ assertTrue(radioController.configWrites.isEmpty())
+ }
+
+ @Test
+ fun terminalPersistenceFailureCannotSuppressHomeRestoration() = runTest {
+ val engine = createEngine(this)
+ engine.startScan(listOf(ChannelOption.SHORT_FAST), dwellDurationSeconds = 60)
+ assertScanActive(engine)
+ discoveryDao.nextSessionWriteFailure = IllegalStateException("database unavailable")
+
+ engine.stopScan()
+ advanceUntilIdle()
+
+ assertEquals(ChannelOption.LONG_FAST.modemPreset, radioController.lastLocalConfig?.lora?.modem_preset)
+ assertEquals(
+ DiscoveryScanState.Complete(DiscoveryScanState.CompletionOutcome.Cancelled),
+ engine.scanState.value,
+ )
+ assertEquals(DiscoverySessionStatus.STOPPED, discoveryDao.sessions.values.single().completionStatus)
+ }
+
+ @Test
+ fun deviceSwitchCancelsOldProcessRestoreWithoutApplyingItToReplacement() = runTest {
+ val firstDevice = "x:FIRST"
+ val secondDevice = "x:SECOND"
+ selectDevice(firstDevice)
+ val engine = createEngine(this)
+ engine.startScan(listOf(ChannelOption.SHORT_FAST), dwellDurationSeconds = 60)
+ assertScanActive(engine)
+ serviceRepository.setConnectionState(ConnectionState.Disconnected)
+
+ engine.stopScan()
+ engine.reset()
+ selectDevice(secondDevice)
+ val secondDeviceHome = ChannelOption.LONG_SLOW
+ radioConfigRepository.setLocalConfigDirect(
+ LocalConfig(lora = Config.LoRaConfig(use_preset = true, modem_preset = secondDeviceHome.modemPreset)),
+ )
+ serviceRepository.setConnectionState(ConnectionState.Connected)
+ engine.startScan(listOf(ChannelOption.MEDIUM_FAST), dwellDurationSeconds = 60)
+ runCurrent()
+
+ val presets = radioController.configWrites.mapNotNull { it.lora?.modem_preset }
+ assertEquals(ChannelOption.MEDIUM_FAST.modemPreset, presets.last())
+ assertFalse(
+ presets.contains(ChannelOption.LONG_FAST.modemPreset),
+ "the first device's delayed home restore must not retune the replacement device",
+ )
+
+ engine.stopScan()
+ advanceUntilIdle()
+
+ val finalPresets = radioController.configWrites.mapNotNull { it.lora?.modem_preset }
+ assertFalse(
+ finalPresets.contains(ChannelOption.LONG_FAST.modemPreset),
+ "the first device's restore must stay rejected after the deferred attempt runs",
+ )
+ assertEquals(
+ secondDeviceHome.modemPreset,
+ finalPresets.last(),
+ "the replacement device must restore its own captured home preset",
+ )
+ }
+
@Test
fun normalCompletionRestoresHomePreset() = runTest {
val engine = createEngine(this)
@@ -676,7 +1033,7 @@ class DiscoveryScanEngineTest {
val state = engine.scanState.value
assertTrue(state is DiscoveryScanState.Complete, "expected Complete, was $state")
assertEquals(DiscoveryScanState.CompletionOutcome.Success, (state as DiscoveryScanState.Complete).outcome)
- assertEquals("complete", discoveryDao.sessions.values.first().completionStatus)
+ assertEquals(DiscoverySessionStatus.COMPLETE, discoveryDao.sessions.values.first().completionStatus)
assertEquals(ChannelOption.LONG_FAST.modemPreset, radioController.lastLocalConfig?.lora?.modem_preset)
}
@@ -685,15 +1042,15 @@ class DiscoveryScanEngineTest {
// region Interrupted-session detection and restore (crash / BLE-loss recovery)
/** Seeds a prior-process interrupted session directly into the fake DAO (keyed by its own id). */
- private fun seedInterruptedSession(
+ private suspend fun seedInterruptedSession(
id: Long = 1L,
deviceAddress: String,
homePreset: String = "LONG_FAST",
- completionStatus: String = "in_progress",
+ completionStatus: String = DiscoverySessionStatus.IN_PROGRESS,
homeLoraConfig: Config.LoRaConfig? =
Config.LoRaConfig(use_preset = true, modem_preset = ChannelOption.LONG_FAST.modemPreset),
) {
- discoveryDao.sessions[id] =
+ discoveryDao.seedSession(
DiscoverySessionEntity(
id = id,
timestamp = 1L,
@@ -702,12 +1059,13 @@ class DiscoveryScanEngineTest {
completionStatus = completionStatus,
deviceAddress = deviceAddress,
homeLoraConfig = homeLoraConfig,
- )
+ ),
+ )
}
@Test
fun startScanCapturesDeviceAddressAndHomeConfigForLaterRestore() = runTest {
- meshPrefs.setDeviceAddress("x:AA:BB:CC:DD:EE:FF")
+ selectDevice("x:AA:BB:CC:DD:EE:FF")
val engine = createEngine(this)
engine.startScan(testPresets, dwellDurationSeconds = 10)
@@ -721,7 +1079,7 @@ class DiscoveryScanEngineTest {
@Test
fun restoreInterruptedSessionsOnReconnectRestoresMatchingDevice() = runTest {
val address = "x:AA:BB:CC:DD:EE:FF"
- meshPrefs.setDeviceAddress(address)
+ selectDevice(address)
// A previous process's scan died mid-flight and never reached restoreHomePreset/finalizeSession.
seedInterruptedSession(deviceAddress = address)
@@ -731,7 +1089,7 @@ class DiscoveryScanEngineTest {
advanceUntilIdle() // serviceRepository is already Connected, so the watcher's first collection fires now
assertEquals(ChannelOption.LONG_FAST.modemPreset, radioController.lastLocalConfig?.lora?.modem_preset)
- assertEquals("restored", discoveryDao.sessions.getValue(1).completionStatus)
+ assertEquals(DiscoverySessionStatus.RESTORED, discoveryDao.sessions.getValue(1).completionStatus)
assertEquals(listOf("LONG_FAST"), restored, "Caller should be notified with the restored home preset")
watcherJob.cancel()
@@ -739,7 +1097,7 @@ class DiscoveryScanEngineTest {
@Test
fun restoreInterruptedSessionsOnReconnectIgnoresDifferentDevice() = runTest {
- meshPrefs.setDeviceAddress("x:CURRENT")
+ selectDevice("x:CURRENT")
seedInterruptedSession(deviceAddress = "x:OTHER-DEVICE")
val restored = mutableListOf<String>()
@@ -748,7 +1106,7 @@ class DiscoveryScanEngineTest {
advanceUntilIdle()
assertNull(radioController.lastLocalConfig, "Should not touch the radio for a session from another device")
- assertEquals("in_progress", discoveryDao.sessions.getValue(1).completionStatus)
+ assertEquals(DiscoverySessionStatus.IN_PROGRESS, discoveryDao.sessions.getValue(1).completionStatus)
assertTrue(restored.isEmpty(), "No notification for a session from another device")
watcherJob.cancel()
@@ -757,7 +1115,7 @@ class DiscoveryScanEngineTest {
@Test
fun restoreInterruptedSessionsOnReconnectSkipsWhileScanActive() = runTest {
val address = "x:AA:BB:CC:DD:EE:FF"
- meshPrefs.setDeviceAddress(address)
+ selectDevice(address)
val engine = createEngine(this)
engine.startScan(testPresets, dwellDurationSeconds = 60)
@@ -775,7 +1133,7 @@ class DiscoveryScanEngineTest {
runCurrent()
assertEquals(
- "in_progress",
+ DiscoverySessionStatus.IN_PROGRESS,
discoveryDao.sessions.getValue(999).completionStatus,
"Stale row must not be touched while this engine's own scan is active",
)
@@ -788,39 +1146,101 @@ class DiscoveryScanEngineTest {
@Test
fun restoreWatcherSurvivesWriteFailureAndRetriesOnNextReconnect() = runTest {
val address = "x:AA:BB:CC:DD:EE:FF"
- meshPrefs.setDeviceAddress(address)
+ selectDevice(address)
seedInterruptedSession(deviceAddress = address)
// The link drops right after Connected, so the restore's config write fails.
radioController.throwOnSetLocalConfig = true
+ radioController.onLocalConfigWriteAttempt = {
+ serviceRepository.setConnectionState(ConnectionState.Disconnected)
+ radioController.onLocalConfigWriteAttempt = null
+ }
val restored = mutableListOf<String>()
val engine = createEngine(this)
val watcherJob = launch { engine.restoreInterruptedSessionsOnReconnect { restored += it } }
advanceUntilIdle()
- assertEquals("in_progress", discoveryDao.sessions.getValue(1).completionStatus, "Failed restore must not mark")
+ assertEquals(
+ DiscoverySessionStatus.IN_PROGRESS,
+ discoveryDao.sessions.getValue(1).completionStatus,
+ "Failed restore must not mark the session restored",
+ )
assertTrue(restored.isEmpty(), "No notification when the restore write failed")
// Next reconnect succeeds — the watcher must still be alive to retry. A real reconnect always passes through
// Disconnected first; the intermediate advanceUntilIdle lets the watcher observe the drop, so the return to
// Connected is a genuine transition rather than a conflated no-op the StateFlow would dedupe away.
radioController.throwOnSetLocalConfig = false
- serviceRepository.setConnectionState(ConnectionState.Disconnected)
- advanceUntilIdle()
serviceRepository.setConnectionState(ConnectionState.Connected)
advanceUntilIdle()
- assertEquals("restored", discoveryDao.sessions.getValue(1).completionStatus)
+ assertEquals(DiscoverySessionStatus.RESTORED, discoveryDao.sessions.getValue(1).completionStatus)
assertEquals(ChannelOption.LONG_FAST.modemPreset, radioController.lastLocalConfig?.lora?.modem_preset)
assertEquals(listOf("LONG_FAST"), restored, "Caller notified once the retry succeeded")
watcherJob.cancel()
}
+ @Test
+ fun restoreWatcherPublishesASingleNotificationWhenRestoreCompletesAfterForegroundTimeout() = runTest {
+ val address = "x:LATE-RESTORE"
+ selectDevice(address)
+ seedInterruptedSession(deviceAddress = address)
+ radioController.failLocalConfigWritesRemaining = Int.MAX_VALUE
+ radioController.onLocalConfigWriteAttempt = {
+ serviceRepository.setConnectionState(ConnectionState.Disconnected)
+ }
+ val restored = mutableListOf<String>()
+ val engine = createEngine(this)
+ val watcherJob = launch { engine.restoreInterruptedSessionsOnReconnect { restored += it } }
+
+ runCurrent()
+ advanceTimeBy(DiscoveryHomeRestorer.FOREGROUND_RESTORE_TIMEOUT_MS + 1)
+ runCurrent()
+ assertTrue(restored.isEmpty(), "the foreground timeout must not report an unfinished restore")
+
+ radioController.failLocalConfigWritesRemaining = 0
+ radioController.onLocalConfigWriteAttempt = null
+ serviceRepository.setConnectionState(ConnectionState.Connected)
+ runCurrent()
+ advanceTimeBy(DiscoveryHomeRestorer.POST_RESTORE_SETTLE_DELAY_MS)
+ runCurrent()
+
+ assertEquals(listOf("LONG_FAST"), restored)
+ serviceRepository.setConnectionState(ConnectionState.Disconnected)
+ runCurrent()
+ serviceRepository.setConnectionState(ConnectionState.Connected)
+ runCurrent()
+ assertEquals(listOf("LONG_FAST"), restored, "late completion must notify exactly once")
+
+ watcherJob.cancel()
+ }
+
+ @Test
+ fun restoreInterruptedSessionsOnReconnectFinalizesPendingCompleteWithoutLegacyRestorePlan() = runTest {
+ val address = "x:AA:BB:CC:DD:EE:FF"
+ selectDevice(address)
+ seedInterruptedSession(
+ deviceAddress = address,
+ completionStatus = DiscoverySessionStatus.RESTORE_PENDING_COMPLETE,
+ homePreset = "CUSTOM",
+ homeLoraConfig = null,
+ )
+
+ val engine = createEngine(this)
+ val watcherJob = launch { engine.restoreInterruptedSessionsOnReconnect {} }
+ advanceUntilIdle()
+
+ assertEquals(DiscoverySessionStatus.COMPLETE, discoveryDao.sessions.getValue(1).completionStatus)
+ assertNull(radioController.lastLocalConfig, "legacy row without a restore plan must not touch the radio")
+
+ watcherJob.cancel()
+ }
+
@Test
fun restoreInterruptedSessionsOnReconnectTerminalizesUnrestorableSession() = runTest {
val address = "x:AA:BB:CC:DD:EE:FF"
- meshPrefs.setDeviceAddress(address)
+ selectDevice(address)
// Config was null at scan start (nothing to restore), so this session can never be recovered.
seedInterruptedSession(deviceAddress = address, homePreset = "CUSTOM", homeLoraConfig = null)
@@ -830,7 +1250,7 @@ class DiscoveryScanEngineTest {
advanceUntilIdle()
// Marked terminal so it stops re-matching getInterruptedSession forever; radio untouched, no notification.
- assertEquals("unrestorable", discoveryDao.sessions.getValue(1).completionStatus)
+ assertEquals(DiscoverySessionStatus.UNRESTORABLE, discoveryDao.sessions.getValue(1).completionStatus)
assertNull(radioController.lastLocalConfig, "Nothing to write to the radio without a captured config")
assertTrue(restored.isEmpty(), "No notification for an unrestorable session")
@@ -838,4 +1258,8 @@ class DiscoveryScanEngineTest {
}
// endregion
+
+ private companion object {
+ const val DEFAULT_DEVICE_ADDRESS = "x:TEST-DEVICE"
+ }
}
diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryTerminalCoordinatorTest.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryTerminalCoordinatorTest.kt
new file mode 100644
index 0000000000..82cae4a963
--- /dev/null
+++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryTerminalCoordinatorTest.kt
@@ -0,0 +1,194 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.feature.discovery
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.CoroutineStart
+import kotlinx.coroutines.async
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import org.meshtastic.core.common.di.ApplicationCoroutineScope
+import org.meshtastic.core.database.entity.DiscoverySessionEntity
+import org.meshtastic.core.database.entity.DiscoverySessionStatus
+import org.meshtastic.core.model.ConnectionState
+import org.meshtastic.core.testing.FakeMeshPrefs
+import org.meshtastic.core.testing.FakeServiceRepository
+import org.meshtastic.proto.Config
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+class DiscoveryTerminalCoordinatorTest {
+ /** Adapts the finite test scope to the application-scope contract for coordinator-only tests. */
+ private fun applicationScope(scope: CoroutineScope): ApplicationCoroutineScope =
+ object : ApplicationCoroutineScope {
+ override val coroutineContext = scope.coroutineContext
+ }
+
+ @Test
+ @Suppress("LongMethod")
+ fun joinedRestoreFailurePublishesTheDowngradedOutcome() = runTest {
+ val device = "x:DEVICE"
+ val discoveryDao = SharedInMemoryDiscoveryDao()
+ val sessionId =
+ discoveryDao.insertSession(
+ DiscoverySessionEntity(
+ timestamp = 1L,
+ presetsScanned = "SHORT_FAST",
+ homePreset = "LONG_FAST",
+ completionStatus = DiscoverySessionStatus.IN_PROGRESS,
+ deviceAddress = device,
+ ),
+ )
+ // Model a device switch so the shared restore resolves false immediately; retry policy is tested separately.
+ val meshPrefs = FakeMeshPrefs().apply { setDeviceAddress("x:OTHER") }
+ val serviceRepository = FakeServiceRepository().apply { setConnectionState(ConnectionState.Connected) }
+ val appScope = applicationScope(this)
+ val radioController = DiscoveryTestRadioController().apply { selectedDeviceAddress = device }
+ val homeRestorer =
+ DiscoveryHomeRestorer(
+ radioController = radioController,
+ serviceRepository = serviceRepository,
+ discoveryDao = discoveryDao,
+ applicationScope = appScope,
+ meshPrefs = meshPrefs,
+ )
+ val publishedOutcomes = mutableListOf<DiscoveryScanState.CompletionOutcome>()
+ val coordinator =
+ DiscoveryTerminalCoordinator(
+ discoveryDao = discoveryDao,
+ homeRestorer = homeRestorer,
+ applicationScope = appScope,
+ onSessionUpdated = {},
+ onTerminalCompleted = { publishedOutcomes += it },
+ cancelScan = {},
+ )
+ val restorePlan =
+ DiscoveryHomeRestorePlan(
+ sessionId = sessionId,
+ deviceAddress = device,
+ loraConfig = Config.LoRaConfig(use_preset = true),
+ primaryChannel = null,
+ restorePrimaryChannel = false,
+ finalStatus = DiscoverySessionStatus.STOPPED,
+ )
+ val ownerRequest =
+ DiscoveryTerminalRequest(
+ sessionId = sessionId,
+ restorePlan = restorePlan,
+ pendingStatus = DiscoverySessionStatus.RESTORE_PENDING_STOPPED,
+ outcome = DiscoveryScanState.CompletionOutcome.Cancelled,
+ awaitRestore = false,
+ shouldGenerateAi = false,
+ )
+ val joinedRequest =
+ ownerRequest.copy(
+ pendingStatus = DiscoverySessionStatus.RESTORE_PENDING_COMPLETE,
+ outcome = DiscoveryScanState.CompletionOutcome.Success,
+ awaitRestore = true,
+ )
+
+ val owner = async(start = CoroutineStart.UNDISPATCHED) { coordinator.complete(ownerRequest) }
+ val joined = async(start = CoroutineStart.UNDISPATCHED) { coordinator.complete(joinedRequest) }
+ runCurrent()
+
+ assertTrue(owner.isCompleted, "terminal owner did not complete")
+ assertEquals(DiscoveryScanState.CompletionOutcome.Cancelled, owner.await())
+ assertTrue(joined.isCompleted, "joined restore follow-up did not complete")
+ assertEquals(DiscoveryScanState.CompletionOutcome.Failed, joined.await())
+ assertEquals(
+ listOf(DiscoveryScanState.CompletionOutcome.Cancelled, DiscoveryScanState.CompletionOutcome.Failed),
+ publishedOutcomes,
+ )
+ assertEquals(
+ DiscoverySessionStatus.RESTORE_PENDING_FAILED,
+ discoveryDao.getSession(sessionId)?.completionStatus,
+ )
+ }
+
+ @Test
+ @Suppress("LongMethod")
+ fun joinedOutcomeDoesNotOverwriteStateAfterTerminalReset() = runTest {
+ val discoveryDao = SharedInMemoryDiscoveryDao()
+ val appScope = applicationScope(this)
+ val sessionId =
+ discoveryDao.insertSession(
+ DiscoverySessionEntity(
+ timestamp = 1L,
+ presetsScanned = "SHORT_FAST",
+ homePreset = "LONG_FAST",
+ completionStatus = DiscoverySessionStatus.IN_PROGRESS,
+ ),
+ )
+ val homeRestorer =
+ DiscoveryHomeRestorer(
+ radioController = DiscoveryTestRadioController(),
+ serviceRepository = FakeServiceRepository(),
+ discoveryDao = discoveryDao,
+ applicationScope = appScope,
+ meshPrefs = FakeMeshPrefs(),
+ )
+ val publishedOutcomes = mutableListOf<DiscoveryScanState.CompletionOutcome>()
+ val coordinator =
+ DiscoveryTerminalCoordinator(
+ discoveryDao = discoveryDao,
+ homeRestorer = homeRestorer,
+ applicationScope = appScope,
+ onSessionUpdated = {},
+ onTerminalCompleted = { publishedOutcomes += it },
+ cancelScan = {},
+ )
+ val joinedEntered = CompletableDeferred<Unit>()
+ val releaseJoined = CompletableDeferred<Unit>()
+ val ownerRequest =
+ DiscoveryTerminalRequest(
+ sessionId = sessionId,
+ restorePlan = null,
+ pendingStatus = DiscoverySessionStatus.COMPLETE,
+ outcome = DiscoveryScanState.CompletionOutcome.Success,
+ awaitRestore = false,
+ shouldGenerateAi = false,
+ )
+ val joinedRequest = ownerRequest.copy(pendingStatus = DiscoverySessionStatus.FAILED)
+
+ val owner = async(start = CoroutineStart.UNDISPATCHED) { coordinator.complete(ownerRequest) }
+ val joined =
+ async(start = CoroutineStart.UNDISPATCHED) {
+ coordinator.complete(
+ joinedRequest,
+ beforeFinalize = {
+ joinedEntered.complete(Unit)
+ releaseJoined.await()
+ error("database unavailable")
+ },
+ )
+ }
+ runCurrent()
+
+ assertTrue(joinedEntered.isCompleted, "joined follow-up did not enter after owner completion")
+ assertTrue(owner.isCompleted, "terminal owner did not complete before reset")
+ assertEquals(DiscoveryScanState.CompletionOutcome.Success, owner.await())
+ assertTrue(coordinator.resetForScan())
+ releaseJoined.complete(Unit)
+ runCurrent()
+
+ assertTrue(joined.isCompleted, "joined follow-up did not complete after release")
+ assertEquals(DiscoveryScanState.CompletionOutcome.Failed, joined.await())
+ assertEquals(listOf(DiscoveryScanState.CompletionOutcome.Success), publishedOutcomes)
+ }
+}
diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryTestRadioController.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryTestRadioController.kt
new file mode 100644
index 0000000000..86b1ec1082
--- /dev/null
+++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryTestRadioController.kt
@@ -0,0 +1,112 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.feature.discovery
+
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import org.meshtastic.core.repository.RadioController
+import org.meshtastic.core.testing.FakeRadioController
+import org.meshtastic.proto.Channel
+import org.meshtastic.proto.Config
+
+/** Discovery-local radio fake with deterministic transient local-config write failures. */
+internal class DiscoveryTestRadioController(private val delegate: FakeRadioController = FakeRadioController()) :
+ RadioController by delegate {
+ private val deviceSwitchMutex = Mutex()
+
+ val configWrites: List<Config>
+ get() = delegate.localConfigs
+
+ val channelWrites: List<Channel>
+ get() = delegate.localChannels
+
+ val lastLocalConfig: Config?
+ get() = delegate.lastLocalConfig
+
+ var throwOnSetLocalConfig: Boolean
+ get() = delegate.throwOnSetLocalConfig
+ set(value) {
+ delegate.throwOnSetLocalConfig = value
+ }
+
+ var requestNeighborInfoFailure: Exception? = null
+ val neighborInfoRequests = mutableListOf<Pair<Int, Int>>()
+
+ var selectedDeviceAddress: String?
+ get() = delegate.selectedDeviceAddress
+ set(value) {
+ delegate.selectedDeviceAddress = value
+ }
+
+ fun setSessionGeneration(generation: Long) {
+ delegate.setSessionGeneration(generation)
+ }
+
+ var failLocalConfigWritesRemaining: Int = 0
+ var failChannelWriteAfter: Int?
+ get() = delegate.failChannelWriteAfter
+ set(value) {
+ delegate.failChannelWriteAfter = value
+ }
+
+ /**
+ * Invoked on each local-config write attempt. Conditional restore holds a non-reentrant device-selection mutex, so
+ * this hook must not call [setDeviceAddress] or [restoreLocalConfiguration].
+ */
+ var onLocalConfigWriteAttempt: (suspend () -> Unit)? = null
+ var localConfigWriteAttempts: Int = 0
+ private set
+
+ var acceptConditionalRestore: Boolean = true
+
+ override suspend fun setLocalConfig(config: Config) {
+ localConfigWriteAttempts++
+ onLocalConfigWriteAttempt?.invoke()
+ if (failLocalConfigWritesRemaining > 0) {
+ failLocalConfigWritesRemaining--
+ error("Fake local config write failure")
+ }
+ delegate.setLocalConfig(config)
+ }
+
+ override suspend fun restoreLocalConfiguration(
+ expectedDeviceAddress: String?,
+ config: Config,
+ primaryChannel: Channel?,
+ ): Boolean = deviceSwitchMutex.withLock {
+ if (
+ expectedDeviceAddress == null ||
+ selectedDeviceAddress != expectedDeviceAddress ||
+ !acceptConditionalRestore
+ ) {
+ return@withLock false
+ }
+ primaryChannel?.let { channel -> delegate.setLocalChannel(channel) }
+ setLocalConfig(config)
+ true
+ }
+
+ override suspend fun requestNeighborInfo(requestId: Int, destNum: Int) {
+ neighborInfoRequests += requestId to destNum
+ requestNeighborInfoFailure?.let { throw it }
+ delegate.requestNeighborInfo(requestId, destNum)
+ }
+
+ override suspend fun setDeviceAddress(address: String) {
+ deviceSwitchMutex.withLock { delegate.setDeviceAddress(address) }
+ }
+}
diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryTestRadioControllerTest.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryTestRadioControllerTest.kt
new file mode 100644
index 0000000000..6f02e3f8da
--- /dev/null
+++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/DiscoveryTestRadioControllerTest.kt
@@ -0,0 +1,85 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.feature.discovery
+
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.async
+import kotlinx.coroutines.test.runCurrent
+import kotlinx.coroutines.test.runTest
+import org.meshtastic.proto.Config
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+class DiscoveryTestRadioControllerTest {
+ @Test
+ fun restoreLocalConfigurationRejectsMissingExpectedDeviceOwnership() = runTest {
+ val controller = DiscoveryTestRadioController()
+
+ val restored =
+ controller.restoreLocalConfiguration(
+ expectedDeviceAddress = null,
+ config = Config(lora = Config.LoRaConfig(use_preset = true)),
+ primaryChannel = null,
+ )
+
+ assertFalse(restored)
+ assertTrue(controller.configWrites.isEmpty())
+ assertTrue(controller.channelWrites.isEmpty())
+ }
+
+ @Test
+ fun requestNeighborInfoPropagatesConfiguredFailure() = runTest {
+ val controller = DiscoveryTestRadioController()
+ controller.requestNeighborInfoFailure = IllegalStateException("Neighbor info failed")
+
+ assertFailsWith<IllegalStateException> { controller.requestNeighborInfo(requestId = 1, destNum = 2) }
+ }
+
+ @Test
+ fun restoreLocalConfigurationSerializesDeviceSelection() = runTest {
+ val controller = DiscoveryTestRadioController()
+ controller.setDeviceAddress("x:FIRST")
+ val writeEntered = CompletableDeferred<Unit>()
+ val releaseWrite = CompletableDeferred<Unit>()
+ controller.onLocalConfigWriteAttempt = {
+ writeEntered.complete(Unit)
+ releaseWrite.await()
+ }
+ val config = Config(lora = Config.LoRaConfig(use_preset = true))
+
+ val restore = async {
+ controller.restoreLocalConfiguration(
+ expectedDeviceAddress = "x:FIRST",
+ config = config,
+ primaryChannel = null,
+ )
+ }
+ writeEntered.await()
+ val selection = async { controller.setDeviceAddress("x:SECOND") }
+ runCurrent()
+
+ assertFalse(selection.isCompleted, "device selection must wait for the in-flight restoration")
+ releaseWrite.complete(Unit)
+ assertTrue(restore.await())
+ selection.await()
+ assertEquals(listOf(config), controller.configWrites)
+ assertEquals("x:SECOND", controller.selectedDeviceAddress)
+ }
+}
diff --git a/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/SharedInMemoryDiscoveryDao.kt b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/SharedInMemoryDiscoveryDao.kt
new file mode 100644
index 0000000000..913fb67f24
--- /dev/null
+++ b/feature/discovery/src/commonTest/kotlin/org/meshtastic/feature/discovery/SharedInMemoryDiscoveryDao.kt
@@ -0,0 +1,262 @@
+/*
+ * Copyright (c) 2026 Meshtastic LLC
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+package org.meshtastic.feature.discovery
+
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import org.meshtastic.core.database.dao.DiscoveryDao
+import org.meshtastic.core.database.entity.DiscoveredNodeEntity
+import org.meshtastic.core.database.entity.DiscoveryPresetResultEntity
+import org.meshtastic.core.database.entity.DiscoverySessionEntity
+import org.meshtastic.core.database.entity.DiscoverySessionStatus
+
+/** Shared in-memory [DiscoveryDao] for discovery feature tests, including live session flows. */
+internal class SharedInMemoryDiscoveryDao : DiscoveryDao {
+ private val stateLock = Mutex()
+ private var nextSessionId = 1L
+ private var nextPresetResultId = 1L
+ private var nextNodeId = 1L
+ private val mutableSessions = mutableMapOf<Long, DiscoverySessionEntity>()
+ private val mutablePresetResults = mutableMapOf<Long, DiscoveryPresetResultEntity>()
+ private val mutableDiscoveredNodes = mutableMapOf<Long, DiscoveredNodeEntity>()
+ private val sessionsFlow = MutableStateFlow<List<DiscoverySessionEntity>>(emptyList())
+ private val presetResultsFlow = MutableStateFlow<List<DiscoveryPresetResultEntity>>(emptyList())
+ private val discoveredNodesFlow = MutableStateFlow<List<DiscoveredNodeEntity>>(emptyList())
+
+ /** Read-only views used by feature-test assertions. */
+ val sessions: Map<Long, DiscoverySessionEntity>
+ get() = sessionsFlow.value.associateBy { it.id }
+
+ val presetResults: Map<Long, DiscoveryPresetResultEntity>
+ get() = presetResultsFlow.value.associateBy { it.id }
+
+ val discoveredNodes: Map<Long, DiscoveredNodeEntity>
+ get() = discoveredNodesFlow.value.associateBy { it.id }
+
+ /** Seeds a persisted session with a stable ID for recovery tests. */
+ suspend fun seedSession(session: DiscoverySessionEntity) {
+ require(session.id > 0) { "seeded discovery sessions require a persisted id" }
+ stateLock.withLock {
+ mutableSessions[session.id] = session
+ nextSessionId = maxOf(nextSessionId, session.id + 1)
+ }
+ refreshSessionsFlow()
+ }
+
+ private suspend fun refreshSessionsFlow() {
+ stateLock.withLock { sessionsFlow.value = mutableSessions.values.sortedByDescending { it.timestamp } }
+ }
+
+ private suspend fun refreshPresetResultsFlow() {
+ stateLock.withLock { presetResultsFlow.value = mutablePresetResults.values.toList() }
+ }
+
+ private suspend fun refreshDiscoveredNodesFlow() {
+ stateLock.withLock { discoveredNodesFlow.value = mutableDiscoveredNodes.values.toList() }
+ }
+
+ override suspend fun insertSession(session: DiscoverySessionEntity): Long {
+ val id =
+ stateLock.withLock {
+ val id = nextSessionId++
+ mutableSessions[id] = session.copy(id = id)
+ id
+ }
+ refreshSessionsFlow()
+ return id
+ }
+
+ override suspend fun updateSession(session: DiscoverySessionEntity) {
+ stateLock.withLock { if (mutableSessions.containsKey(session.id)) mutableSessions[session.id] = session }
+ refreshSessionsFlow()
+ }
+
+ override fun getAllSessions(): Flow<List<DiscoverySessionEntity>> = sessionsFlow
+
+ // The Room snapshot query intentionally has no ORDER BY; preserve that contract instead of mirroring the live flow.
+ override suspend fun getAllSessionsSnapshot(): List<DiscoverySessionEntity> =
+ stateLock.withLock { mutableSessions.values.toList() }
+
+ override suspend fun getSession(sessionId: Long): DiscoverySessionEntity? =
+ stateLock.withLock { mutableSessions[sessionId] }
+
+ override suspend fun updateSessionCompletionStatus(sessionId: Long, status: String): Int {
+ val updated =
+ stateLock.withLock {
+ val session = mutableSessions[sessionId] ?: return@withLock false
+ mutableSessions[sessionId] = session.copy(completionStatus = status)
+ true
+ }
+ if (updated) refreshSessionsFlow()
+ return if (updated) 1 else 0
+ }
+
+ override suspend fun updateSessionAggregates(
+ sessionId: Long,
+ totalUniqueNodes: Int,
+ totalDwellSeconds: Long,
+ totalMessages: Int,
+ totalSensorPackets: Int,
+ furthestNodeDistance: Double,
+ avgChannelUtilization: Double,
+ ) {
+ val updated =
+ stateLock.withLock {
+ val session = mutableSessions[sessionId] ?: return@withLock false
+ mutableSessions[sessionId] =
+ session.copy(
+ totalUniqueNodes = totalUniqueNodes,
+ totalDwellSeconds = totalDwellSeconds,
+ totalMessages = totalMessages,
+ totalSensorPackets = totalSensorPackets,
+ furthestNodeDistance = furthestNodeDistance,
+ avgChannelUtilization = avgChannelUtilization,
+ )
+ true
+ }
+ if (updated) refreshSessionsFlow()
+ }
+
+ override suspend fun updateRecoverableSessionCompletionStatus(sessionId: Long, status: String): Int {
+ val updated =
+ stateLock.withLock {
+ val session =
+ mutableSessions[sessionId]?.takeIf { it.completionStatus in DiscoverySessionStatus.RECOVERABLE }
+ ?: return@withLock false
+ mutableSessions[sessionId] = session.copy(completionStatus = status)
+ true
+ }
+ if (updated) refreshSessionsFlow()
+ return if (updated) 1 else 0
+ }
+
+ override fun getSessionFlow(sessionId: Long): Flow<DiscoverySessionEntity?> =
+ sessionsFlow.map { sessions -> sessions.firstOrNull { it.id == sessionId } }
+
+ override suspend fun deleteSession(sessionId: Long) {
+ stateLock.withLock {
+ mutableSessions.remove(sessionId)
+ val resultIds = mutablePresetResults.values.filter { it.sessionId == sessionId }.map { it.id }
+ resultIds.forEach { resultId ->
+ mutableDiscoveredNodes.entries.removeAll { it.value.presetResultId == resultId }
+ mutablePresetResults.remove(resultId)
+ }
+ }
+ refreshSessionsFlow()
+ refreshPresetResultsFlow()
+ refreshDiscoveredNodesFlow()
+ }
+
+ override suspend fun insertPresetResult(result: DiscoveryPresetResultEntity): Long {
+ val id =
+ stateLock.withLock {
+ val id = nextPresetResultId++
+ mutablePresetResults[id] = result.copy(id = id)
+ id
+ }
+ refreshPresetResultsFlow()
+ return id
+ }
+
+ override suspend fun updatePresetResult(result: DiscoveryPresetResultEntity) {
+ stateLock.withLock { if (mutablePresetResults.containsKey(result.id)) mutablePresetResults[result.id] = result }
+ refreshPresetResultsFlow()
+ }
+
+ override suspend fun getPresetResults(sessionId: Long): List<DiscoveryPresetResultEntity> =
+ stateLock.withLock { mutablePresetResults.values.filter { it.sessionId == sessionId } }
+
+ override fun getPresetResultsFlow(sessionId: Long): Flow<List<DiscoveryPresetResultEntity>> =
+ presetResultsFlow.map { results -> results.filter { it.sessionId == sessionId } }
+
+ override suspend fun insertDiscoveredNode(node: DiscoveredNodeEntity): Long {
+ val id =
+ stateLock.withLock {
+ val id = nextNodeId++
+ mutableDiscoveredNodes[id] = node.copy(id = id)
+ id
+ }
+ refreshDiscoveredNodesFlow()
+ return id
+ }
+
+ override suspend fun insertDiscoveredNodes(nodes: List<DiscoveredNodeEntity>) {
+ stateLock.withLock {
+ nodes.forEach { node ->
+ val id = nextNodeId++
+ mutableDiscoveredNodes[id] = node.copy(id = id)
+ }
+ }
+ refreshDiscoveredNodesFlow()
+ }
+
+ override suspend fun updateDiscoveredNode(node: DiscoveredNodeEntity) {
+ stateLock.withLock { if (mutableDiscoveredNodes.containsKey(node.id)) mutableDiscoveredNodes[node.id] = node }
+ refreshDiscoveredNodesFlow()
+ }
+
+ override suspend fun getDiscoveredNodes(presetResultId: Long): List<DiscoveredNodeEntity> =
+ stateLock.withLock { mutableDiscoveredNodes.values.filter { it.presetResultId == presetResultId } }
+
+ override fun getDiscoveredNodesFlow(presetResultId: Long): Flow<List<DiscoveredNodeEntity>> =
+ discoveredNodesFlow.map { nodes -> nodes.filter { it.presetResultId == presetResultId } }
+
+ override suspend fun getUniqueNodeNums(sessionId: Long): List<Long> = stateLock.withLock {
+ mutablePresetResults.values
+ .filter { it.sessionId == sessionId }
+ .flatMap { result -> mutableDiscoveredNodes.values.filter { it.presetResultId == result.id } }
+ .map { it.nodeNum }
+ .distinct()
+ }
+
+ override suspend fun getUniqueNodeCount(sessionId: Long): Int = getUniqueNodeNums(sessionId).size
+
+ override suspend fun getMaxDistance(sessionId: Long): Double? = stateLock.withLock {
+ mutablePresetResults.values
+ .filter { it.sessionId == sessionId }
+ .flatMap { result -> mutableDiscoveredNodes.values.filter { it.presetResultId == result.id } }
+ .mapNotNull { it.distanceFromUser }
+ .maxOrNull()
+ }
+
+ // The DAO method name is historical; the Room query currently returns only the session entity.
+ override suspend fun getSessionWithResults(sessionId: Long): DiscoverySessionEntity? =
+ stateLock.withLock { mutableSessions[sessionId] }
+
+ override suspend fun markInterruptedSessions() {
+ stateLock.withLock {
+ mutableSessions.keys.toList().forEach { key ->
+ val session = checkNotNull(mutableSessions[key])
+ if (session.completionStatus == DiscoverySessionStatus.IN_PROGRESS) {
+ mutableSessions[key] = session.copy(completionStatus = DiscoverySessionStatus.INTERRUPTED)
+ }
+ }
+ }
+ refreshSessionsFlow()
+ }
+
+ override suspend fun getInterruptedSession(deviceAddress: String): DiscoverySessionEntity? = stateLock.withLock {
+ mutableSessions.values
+ .filter {
+ it.deviceAddress == deviceAddress && it.completionStatus in DiscoverySessionStatus.RECOVERABLE
+ }
+ .maxByOrNull { it.timestamp }
+ }
+}
Served by rngit 1.5.0 - Generated in 0.34s